MINOR: Cleanup metadata module (#18937)

Removed unused code and fixed IDEA warnings.

Reviewers: Chia-Ping Tsai <[email protected]>

Author: wernerdv

core/src/main/scala/kafka/server/metadata/ScramPublisher.scala

@@ -56,7 +56,7 @@ class ScramPublisher(
             userChanges.forEach {
               case (userName, change) =>
                 if (change.isPresent) {
-                  credentialProvider.updateCredential(mechanism, userName, change.get().toCredential(mechanism))
+                  credentialProvider.updateCredential(mechanism, userName, change.get().toCredential)
                 } else {
                   credentialProvider.removeCredentials(mechanism, userName)
                 }

metadata/src/main/java/org/apache/kafka/controller/BrokersToIsrs.java

@@ -278,10 +278,6 @@ PartitionsOnReplicaIterator partitionsWithNoLeader() {
         return iterator(NO_LEADER, true);
     }
 
-    PartitionsOnReplicaIterator partitionsLedByBroker(int brokerId) {
-        return iterator(brokerId, true);
-    }
-
     PartitionsOnReplicaIterator partitionsWithBrokerInIsr(int brokerId) {
         return iterator(brokerId, false);
     }

metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java

@@ -521,7 +521,7 @@ BrokerFeature processRegistrationFeature(
      * @param brokerId      The broker id to track.
      * @param brokerEpoch   The broker epoch to track.
      *
-     * @returns             True only if the ClusterControlManager is active.
+     * @return              True only if the ClusterControlManager is active.
      */
     boolean trackBrokerHeartbeat(int brokerId, long brokerEpoch) {
         BrokerHeartbeatManager manager = heartbeatManager;

metadata/src/main/java/org/apache/kafka/controller/OffsetControlManager.java

@@ -418,9 +418,4 @@ public void replay(AbortTransactionRecord message, long offset) {
         log.info("Replayed {} at offset {}. Reverted to offset {}.",
                 message, offset, preTransactionOffset);
     }
-
-    // VisibleForTesting
-    void setNextWriteOffset(long newNextWriteOffset) {
-        this.nextWriteOffset = newNextWriteOffset;
-    }
 }

metadata/src/main/java/org/apache/kafka/controller/QuorumController.java

@@ -177,11 +177,6 @@ public final class QuorumController implements Controller {
      */
     private static final int DEFAULT_MAX_RECORDS_PER_BATCH = 10000;
 
-    /**
-     * The default minimum event time that can be logged as a slow event.
-     */
-    private static final int DEFAULT_MIN_SLOW_EVENT_TIME_MS = 200;
-
     /**
      * The maximum records any user-initiated operation is allowed to generate.
      *
@@ -677,11 +672,6 @@ public String toString() {
         }
     }
 
-    // Visible for testing
-    OffsetControlManager offsetControl() {
-        return offsetControl;
-    }
-
     // Visible for testing
     ReplicationControlManager replicationControl() {
         return replicationControl;

metadata/src/main/java/org/apache/kafka/controller/QuorumFeatures.java

@@ -84,10 +84,6 @@ public int nodeId() {
         return nodeId;
     }
 
-    public Map<String, VersionRange> localSupportedFeatures() {
-        return localSupportedFeatures;
-    }
-
     public List<Integer> quorumNodeIds() {
         return quorumNodeIds;
     }

metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java

@@ -2071,9 +2071,8 @@ void generateLeaderAndIsrUpdates(String context,
                     alterPartitionReassignment(topic.name(), partition, records, allowRFChange);
                     successfulAlterations++;
                 } catch (Throwable e) {
-                    log.info("Unable to alter partition reassignment for " +
-                        topic.name() + ":" + partition.partitionIndex() + " because " +
-                        "of an " + e.getClass().getSimpleName() + " error: " + e.getMessage());
+                    log.info("Unable to alter partition reassignment for {}:{} because of an {} error: {}",
+                            topic.name(), partition.partitionIndex(), e.getClass().getSimpleName(), e.getMessage());
                     error = ApiError.fromThrowable(e);
                 }
                 totalAlterations++;

metadata/src/main/java/org/apache/kafka/controller/errors/ControllerExceptions.java

@@ -40,21 +40,6 @@ public static boolean isTimeoutException(Throwable exception) {
         return exception instanceof TimeoutException;
     }
 
-    /**
-     * Check if an exception is a NotController exception.
-     *
-     * @param exception     The exception to check.
-     * @return              True if the exception is a NotController exception.
-     */
-    public static boolean isNotControllerException(Throwable exception) {
-        if (exception == null) return false;
-        if (exception instanceof ExecutionException) {
-            exception = exception.getCause();
-            if (exception == null) return false;
-        }
-        return exception instanceof NotControllerException;
-    }
-
     /**
      * Create a new exception indicating that the controller is in pre-migration mode, so the
      * operation cannot be completed.

metadata/src/main/java/org/apache/kafka/image/ClusterImage.java

@@ -66,10 +66,6 @@ public Map<Integer, ControllerRegistration> controllers() {
         return controllers;
     }
 
-    public boolean containsBroker(int brokerId) {
-        return brokers.containsKey(brokerId);
-    }
-
     public long brokerEpoch(int brokerId) {
         BrokerRegistration brokerRegistration = broker(brokerId);
         if (brokerRegistration == null) {

metadata/src/main/java/org/apache/kafka/image/TopicDelta.java

@@ -129,20 +129,6 @@ public TopicImage apply() {
         return new TopicImage(image.name(), image.id(), newPartitions);
     }
 
-    public boolean hasPartitionsWithAssignmentChanges() {
-        for (Entry<Integer, PartitionRegistration> entry : partitionChanges.entrySet()) {
-            int partitionId = entry.getKey();
-            // New Partition.
-            if (!image.partitions().containsKey(partitionId))
-                return true;
-            PartitionRegistration previousPartition = image.partitions().get(partitionId);
-            PartitionRegistration currentPartition = entry.getValue();
-            if (!previousPartition.hasSameAssignment(currentPartition))
-                return true;
-        }
-        return false;
-    }
-
     /**
      * Find the partitions that have change based on the replica given.
      * <p>

metadata/src/main/java/org/apache/kafka/image/publisher/SnapshotGenerator.java

@@ -23,7 +23,6 @@
 import org.apache.kafka.image.MetadataImage;
 import org.apache.kafka.image.loader.LoaderManifest;
 import org.apache.kafka.image.loader.LogDeltaManifest;
-import org.apache.kafka.image.loader.SnapshotManifest;
 import org.apache.kafka.queue.EventQueue;
 import org.apache.kafka.queue.KafkaEventQueue;
 import org.apache.kafka.server.fault.FaultHandler;
@@ -217,28 +216,20 @@ public void onMetadataUpdate(
     ) {
         switch (manifest.type()) {
             case LOG_DELTA:
-                publishLogDelta(delta, newImage, (LogDeltaManifest) manifest);
+                publishLogDelta(newImage, (LogDeltaManifest) manifest);
                 break;
             case SNAPSHOT:
-                publishSnapshot(delta, newImage, (SnapshotManifest) manifest);
+                publishSnapshot(newImage);
                 break;
         }
     }
 
-    void publishSnapshot(
-        MetadataDelta delta,
-        MetadataImage newImage,
-        SnapshotManifest manifest
-    ) {
+    void publishSnapshot(MetadataImage newImage) {
         log.debug("Resetting the snapshot counters because we just read {}.", newImage.provenance().snapshotName());
         resetSnapshotCounters();
     }
 
-    void publishLogDelta(
-        MetadataDelta delta,
-        MetadataImage newImage,
-        LogDeltaManifest manifest
-    ) {
+    void publishLogDelta(MetadataImage newImage, LogDeltaManifest manifest) {
         bytesSinceLastSnapshot += manifest.numBytes();
         if (bytesSinceLastSnapshot >= maxBytesSinceLastSnapshot) {
             if (eventQueue.isEmpty()) {

metadata/src/main/java/org/apache/kafka/metadata/LeaderAndIsr.java

@@ -26,9 +26,6 @@ public class LeaderAndIsr {
     public static final int INITIAL_LEADER_EPOCH = 0;
     public static final int INITIAL_PARTITION_EPOCH = 0;
     public static final int NO_LEADER = -1;
-    public static final int NO_EPOCH = -1;
-    public static final int LEADER_DURING_DELETE = -2;
-    public static final int EPOCH_DURING_DELETE = -2;
 
     private final int leader;
     private final int leaderEpoch;
@@ -74,10 +71,6 @@ public LeaderAndIsr(
         this.partitionEpoch = partitionEpoch;
     }
 
-    public static LeaderAndIsr duringDelete(List<Integer> isr) {
-        return new LeaderAndIsr(LEADER_DURING_DELETE, isr);
-    }
-
     public int leader() {
         return leader;
     }
@@ -132,20 +125,6 @@ public List<Integer> isr() {
                 .toList();
     }
 
-    public boolean equalsAllowStalePartitionEpoch(LeaderAndIsr other) {
-        if (this == other) {
-            return true;
-        } else if (other == null) {
-            return false;
-        } else {
-            return leader == other.leader &&
-                    leaderEpoch == other.leaderEpoch &&
-                    isrWithBrokerEpoch.equals(other.isrWithBrokerEpoch) &&
-                    leaderRecoveryState == other.leaderRecoveryState &&
-                    partitionEpoch <= other.partitionEpoch;
-        }
-    }
-
     @Override
     public String toString() {
         return "LeaderAndIsr(" +

metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java

@@ -470,11 +470,4 @@ public String toString() {
                 ", partitionEpoch=" + partitionEpoch +
                 ")";
     }
-
-    public boolean hasSameAssignment(PartitionRegistration registration) {
-        return Arrays.equals(this.replicas, registration.replicas) &&
-            Arrays.equals(this.directories, registration.directories) &&
-            Arrays.equals(this.addingReplicas, registration.addingReplicas) &&
-            Arrays.equals(this.removingReplicas, registration.removingReplicas);
-    }
 }

metadata/src/main/java/org/apache/kafka/metadata/ScramCredentialData.java

@@ -86,7 +86,7 @@ public UserScramCredentialRecord toRecord(
                 setIterations(iterations);
     }
 
-    public ScramCredential toCredential(ScramMechanism mechanism) {
+    public ScramCredential toCredential() {
         return new ScramCredential(salt, storedKey, serverKey, iterations);
     }
 

metadata/src/main/java/org/apache/kafka/metadata/VersionRange.java

@@ -56,13 +56,6 @@ public boolean contains(short version) {
         return version >= min && version <= max;
     }
 
-    /**
-     * Check if a given version range has overlap with this one
-     */
-    public boolean intersects(VersionRange other) {
-        return other.min <= max && other.max >= min;
-    }
-
     @Override
     public int hashCode() {
         return Objects.hash(min, max);

metadata/src/main/java/org/apache/kafka/metadata/util/SnapshotFileReader.java

@@ -208,8 +208,4 @@ public void close() throws Exception {
         beginShutdown("closing");
         queue.close();
     }
-
-    public CompletableFuture<Void> caughtUpFuture() {
-        return caughtUpFuture;
-    }
 }

metadata/src/test/java/org/apache/kafka/controller/ConfigurationControlManagerTest.java

@@ -94,8 +94,6 @@ public class ConfigurationControlManagerTest {
     static {
         SYNONYMS.put("abc", List.of(new ConfigSynonym("foo.bar")));
         SYNONYMS.put("def", List.of(new ConfigSynonym("baz")));
-        SYNONYMS.put(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG,
-            List.of(new ConfigSynonym(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG)));
         SYNONYMS.put("quuux", List.of(new ConfigSynonym("quux", HOURS_TO_MILLISECONDS)));
         SYNONYMS.put(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, List.of(new ConfigSynonym(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG)));
     }
@@ -130,7 +128,7 @@ static <A, B> Entry<A, B> entry(A a, B b) {
     }
 
     @Test
-    public void testReplay() throws Exception {
+    public void testReplay() {
         ConfigurationControlManager manager = new ConfigurationControlManager.Builder().
             setKafkaConfigSchema(SCHEMA).
             build();
@@ -314,7 +312,7 @@ public void validate(RequestMetadata actual) throws PolicyViolationException {
         }
 
         @Override
-        public void close() throws Exception {
+        public void close() {
             // nothing to do
         }
 
@@ -379,7 +377,7 @@ public void validate(RequestMetadata actual) throws PolicyViolationException {
         }
 
         @Override
-        public void close() throws Exception {
+        public void close() {
             // empty
         }
 

metadata/src/test/java/org/apache/kafka/controller/ProducerIdControlManagerTest.java

@@ -154,12 +154,11 @@ public void testGenerateProducerIds() {
         assertEquals(new ProducerIdsBlock(3, 100000, 1000), producerIdControlManager.nextProducerBlock());
     }
 
-    static ProducerIdsBlock generateProducerIds(
+    static void generateProducerIds(
             ProducerIdControlManager producerIdControlManager, int brokerId, long brokerEpoch) {
         ControllerResult<ProducerIdsBlock> result =
             producerIdControlManager.generateNextProducerId(brokerId, brokerEpoch);
         result.records().forEach(apiMessageAndVersion ->
             producerIdControlManager.replay((ProducerIdsRecord) apiMessageAndVersion.message()));
-        return result.response();
     }
 }

metadata/src/test/java/org/apache/kafka/controller/QuorumControllerIntegrationTestUtils.java

@@ -53,10 +53,6 @@
 public class QuorumControllerIntegrationTestUtils {
     private static final Logger log = LoggerFactory.getLogger(QuorumControllerIntegrationTestUtils.class);
 
-    BrokerRegistrationRequestData.FeatureCollection brokerFeatures() {
-        return brokerFeatures(MetadataVersion.MINIMUM_VERSION, MetadataVersion.latestTesting());
-    }
-
     /**
      * Create a broker features collection for use in a registration request. We only set MV. here.
      *

metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java

@@ -97,7 +97,6 @@
 import org.apache.kafka.metadata.RecordTestUtils;
 import org.apache.kafka.metadata.RecordTestUtils.ImageDeltaPair;
 import org.apache.kafka.metadata.RecordTestUtils.TestThroughAllIntermediateImagesLeadingToFinalImageHelper;
-import org.apache.kafka.metadata.authorizer.StandardAuthorizer;
 import org.apache.kafka.metadata.bootstrap.BootstrapMetadata;
 import org.apache.kafka.metadata.util.BatchFileWriter;
 import org.apache.kafka.metalog.LocalLogManager;
@@ -144,7 +143,6 @@
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
-import java.util.stream.IntStream;
 
 import static java.util.function.Function.identity;
 import static org.apache.kafka.clients.admin.AlterConfigOp.OpType.SET;
@@ -1409,11 +1407,6 @@ public void testConfigResourceExistenceChecker() throws Throwable {
         }
     }
 
-    private static final Uuid FOO_ID = Uuid.fromString("igRktLOnR8ektWHr79F8mw");
-
-    private static final Map<Integer, Long> ALL_ZERO_BROKER_EPOCHS =
-        IntStream.of(0, 1, 2, 3).boxed().collect(Collectors.toMap(identity(), __ -> 0L));
-
     @Test
     public void testFatalMetadataReplayErrorOnActive() throws Throwable {
         try (
@@ -1476,13 +1469,6 @@ public void testFatalMetadataErrorDuringLogLoading() throws Exception {
         }
     }
 
-    private static void assertInitialLoadFuturesNotComplete(List<StandardAuthorizer> authorizers) {
-        for (int i = 0; i < authorizers.size(); i++) {
-            assertFalse(authorizers.get(i).initialLoadFuture().isDone(),
-                "authorizer " + i + " should not have completed loading.");
-        }
-    }
-
     static class InitialSnapshot implements AutoCloseable {
         File tempDir;
         BatchFileWriter writer;