KAFKA-18999 Remove BrokerMetadata (#19227)

* Replace `BrokerMetadata` with `UsableBroker` in KRaftMetadataCache and
ReassignPartitionsCommand.

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

Author: FrankYang0529

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

@@ -18,7 +18,6 @@
 package kafka.server.metadata
 
 import kafka.utils.Logging
-import org.apache.kafka.admin.BrokerMetadata
 import org.apache.kafka.common._
 import org.apache.kafka.common.config.ConfigResource
 import org.apache.kafka.common.errors.InvalidTopicException
@@ -341,13 +340,6 @@ class KRaftMetadataCache(
     Option(_currentImage.cluster.broker(brokerId)).count(_.inControlledShutdown) == 1
   }
 
-  private def getAliveBrokers(image: MetadataImage): util.List[BrokerMetadata] = {
-    image.cluster().brokers().values().stream()
-      .filter(Predicate.not(_.fenced))
-      .map(broker => new BrokerMetadata(broker.id, broker.rack))
-      .collect(Collectors.toList())
-  }
-
   override def getAliveBrokerNode(brokerId: Int, listenerName: ListenerName): util.Optional[Node] = {
     util.Optional.ofNullable(_currentImage.cluster().broker(brokerId))
       .filter(Predicate.not(_.fenced))
@@ -422,11 +414,13 @@ class KRaftMetadataCache(
   }
 
   private def getRandomAliveBroker(image: MetadataImage): util.Optional[Integer] = {
-    val aliveBrokers = getAliveBrokers(image)
+    val aliveBrokers = image.cluster().brokers().values().stream()
+      .filter(Predicate.not(_.fenced))
+      .map(_.id()).toList
     if (aliveBrokers.isEmpty) {
       util.Optional.empty()
     } else {
-      util.Optional.of(aliveBrokers.get(ThreadLocalRandom.current().nextInt(aliveBrokers.size)).id)
+      util.Optional.of(aliveBrokers.get(ThreadLocalRandom.current().nextInt(aliveBrokers.size)))
     }
   }
 

server-common/src/main/java/org/apache/kafka/admin/BrokerMetadata.java

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.tools.reassign;
 
-import org.apache.kafka.admin.BrokerMetadata;
 import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.admin.AlterConfigOp;
@@ -569,8 +568,8 @@ public static Entry<Map<TopicPartition, List<Integer>>, Map<TopicPartition, List
         List<String> topicsToReassign = t0.getValue();
 
         Map<TopicPartition, List<Integer>> currentAssignments = getReplicaAssignmentForTopics(adminClient, topicsToReassign);
-        List<BrokerMetadata> brokerMetadatas = getBrokerMetadata(adminClient, brokersToReassign, enableRackAwareness);
-        Map<TopicPartition, List<Integer>> proposedAssignments = calculateAssignment(currentAssignments, brokerMetadatas);
+        List<UsableBroker> usableBrokers = getBrokerMetadata(adminClient, brokersToReassign, enableRackAwareness);
+        Map<TopicPartition, List<Integer>> proposedAssignments = calculateAssignment(currentAssignments, usableBrokers);
         System.out.printf("Current partition replica assignment%n%s%n%n",
             formatAsReassignmentJson(currentAssignments, Collections.emptyMap()));
         System.out.printf("Proposed partition reassignment configuration%n%s%n",
@@ -582,12 +581,12 @@ public static Entry<Map<TopicPartition, List<Integer>>, Map<TopicPartition, List
      * Calculate the new partition assignments to suggest in --generate.
      *
      * @param currentAssignment  The current partition assignments.
-     * @param brokerMetadatas    The rack information for each broker.
+     * @param brokers            The rack information for each broker.
      *
      * @return                   A map from partitions to the proposed assignments for each.
      */
     private static Map<TopicPartition, List<Integer>> calculateAssignment(Map<TopicPartition, List<Integer>> currentAssignment,
-                                                                          List<BrokerMetadata> brokerMetadatas) {
+                                                                          List<UsableBroker> usableBrokers) {
         Map<String, List<Entry<TopicPartition, List<Integer>>>> groupedByTopic = new HashMap<>();
         for (Entry<TopicPartition, List<Integer>> e : currentAssignment.entrySet())
             groupedByTopic.computeIfAbsent(e.getKey().topic(), k -> new ArrayList<>()).add(e);
@@ -601,11 +600,7 @@ private static Map<TopicPartition, List<Integer>> calculateAssignment(Map<TopicP
                     new ClusterDescriber() {
                         @Override
                         public Iterator<UsableBroker> usableBrokers() {
-                            return brokerMetadatas.stream().map(brokerMetadata -> new UsableBroker(
-                                    brokerMetadata.id,
-                                    brokerMetadata.rack,
-                                    false
-                            )).iterator();
+                            return usableBrokers.iterator();
                         }
 
                         @Override
@@ -701,16 +696,16 @@ static Map<TopicPartition, List<Integer>> getReplicaAssignmentForPartitions(Admi
      * @return                    The metadata for each broker that was found.
      *                            Brokers that were not found will be omitted.
      */
-    static List<BrokerMetadata> getBrokerMetadata(Admin adminClient, List<Integer> brokers, boolean enableRackAwareness) throws ExecutionException, InterruptedException {
+    static List<UsableBroker> getBrokerMetadata(Admin adminClient, List<Integer> brokers, boolean enableRackAwareness) throws ExecutionException, InterruptedException {
         Set<Integer> brokerSet = new HashSet<>(brokers);
-        List<BrokerMetadata> results = adminClient.describeCluster().nodes().get().stream()
+        List<UsableBroker> results = adminClient.describeCluster().nodes().get().stream()
             .filter(node -> brokerSet.contains(node.id()))
             .map(node -> (enableRackAwareness && node.rack() != null)
-                ? new BrokerMetadata(node.id(), Optional.of(node.rack()))
-                : new BrokerMetadata(node.id(), Optional.empty())
+                ? new UsableBroker(node.id(), Optional.of(node.rack()), false)
+                : new UsableBroker(node.id(), Optional.empty(), false)
             ).collect(Collectors.toList());
 
-        long numRackless = results.stream().filter(m -> m.rack.isEmpty()).count();
+        long numRackless = results.stream().filter(m -> m.rack().isEmpty()).count();
         if (enableRackAwareness && numRackless != 0 && numRackless != results.size()) {
             throw new AdminOperationException("Not all brokers have rack information. Add " +
                 "--disable-rack-aware in command line to make replica assignment without rack " +

tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.tools.reassign;
 
-import org.apache.kafka.admin.BrokerMetadata;
 import org.apache.kafka.clients.admin.Config;
 import org.apache.kafka.clients.admin.MockAdminClient;
 import org.apache.kafka.clients.admin.PartitionReassignment;
@@ -29,6 +28,7 @@
 import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
 import org.apache.kafka.common.utils.Exit;
 import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.metadata.placement.UsableBroker;
 import org.apache.kafka.server.common.AdminCommandFailedException;
 import org.apache.kafka.server.common.AdminOperationException;
 import org.apache.kafka.server.config.QuotaConfig;
@@ -317,19 +317,19 @@ public void testGetBrokerRackInformation() throws Exception {
             build()) {
 
             assertEquals(asList(
-                new BrokerMetadata(0, Optional.of("rack0")),
-                new BrokerMetadata(1, Optional.of("rack1"))
+                new UsableBroker(0, Optional.of("rack0"), false),
+                new UsableBroker(1, Optional.of("rack1"), false)
             ), getBrokerMetadata(adminClient, asList(0, 1), true));
             assertEquals(asList(
-                new BrokerMetadata(0, Optional.empty()),
-                new BrokerMetadata(1, Optional.empty())
+                new UsableBroker(0, Optional.empty(), false),
+                new UsableBroker(1, Optional.empty(), false)
             ), getBrokerMetadata(adminClient, asList(0, 1), false));
             assertStartsWith("Not all brokers have rack information",
                 assertThrows(AdminOperationException.class,
                     () -> getBrokerMetadata(adminClient, asList(1, 2), true)).getMessage());
             assertEquals(asList(
-                new BrokerMetadata(1, Optional.empty()),
-                new BrokerMetadata(2, Optional.empty())
+                new UsableBroker(1, Optional.empty(), false),
+                new UsableBroker(2, Optional.empty(), false)
             ), getBrokerMetadata(adminClient, asList(1, 2), false));
         }
     }