Kafka 2.4 KIP-455 - Update TopicFactory Interface and DefaultTopicFactory Implementation to use AdminClient

Kafka 2.4 KIP-455 - Update TopicFactory Interface and DefaultTopicFactory Implementation to use AdminClient
Signed-off-by: Andrew Choi <[email protected]>

Author: Andrew Choi

src/main/java/com/linkedin/xinfra/monitor/services/ClusterTopicManipulationService.java

@@ -163,7 +163,7 @@ private void createDeleteClusterTopic() {
       try {
         int brokerCount = _adminClient.describeCluster().nodes().get().size();
 
-        Set<Integer> blackListedBrokers = _topicFactory.getBlackListedBrokers(_zkConnect);
+        Set<Integer> blackListedBrokers = _topicFactory.getBlackListedBrokers(_adminClient);
         Set<BrokerMetadata> brokers = new HashSet<>();
         for (Node broker : _adminClient.describeCluster().nodes().get()) {
           BrokerMetadata brokerMetadata = new BrokerMetadata(broker.id(), null);

src/main/java/com/linkedin/xinfra/monitor/services/MultiClusterTopicManagementService.java

@@ -313,7 +313,7 @@ void maybeAddPartitions(int minPartitionNum) throws ExecutionException, Interrup
       if (partitionNum < minPartitionNum) {
         LOGGER.info("{} will increase partition of the topic {} in the cluster from {}" + " to {}.",
             this.getClass().toString(), _topic, partitionNum, minPartitionNum);
-        Set<Integer> blackListedBrokers = _topicFactory.getBlackListedBrokers(_zkConnect);
+        Set<Integer> blackListedBrokers = _topicFactory.getBlackListedBrokers(_adminClient);
         Set<BrokerMetadata> brokers = new HashSet<>();
         for (Node broker : _adminClient.describeCluster().nodes().get()) {
           BrokerMetadata brokerMetadata = new BrokerMetadata(broker.id(), null);
@@ -409,7 +409,7 @@ int numPartitions() throws InterruptedException, ExecutionException {
 
     private Set<Node> getAvailableBrokers() throws ExecutionException, InterruptedException {
       Set<Node> brokers = new HashSet<>(_adminClient.describeCluster().nodes().get());
-      Set<Integer> blackListedBrokers = _topicFactory.getBlackListedBrokers(_zkConnect);
+      Set<Integer> blackListedBrokers = _topicFactory.getBlackListedBrokers(_adminClient);
       brokers.removeIf(broker -> blackListedBrokers.contains(broker.id()));
       return brokers;
     }

src/main/java/com/linkedin/xinfra/monitor/topicfactory/DefaultTopicFactory.java

@@ -32,7 +32,7 @@ public int createTopicIfNotExist(String topic, short replicationFactor, double p
   }
 
   @Override
-  public Set<Integer> getBlackListedBrokers(String zkUrl) {
+  public Set<Integer> getBlackListedBrokers(AdminClient adminClient) {
     return Collections.emptySet();
   }
 }

src/main/java/com/linkedin/xinfra/monitor/topicfactory/TopicFactory.java

@@ -39,9 +39,9 @@ int createTopicIfNotExist(String topic, short replicationFactor, double partitio
       throws ExecutionException, InterruptedException;
 
   /**
-   * @param zkUrl zookeeper connection url
+   * @param adminClient AdminClient object
    * @return A set of brokers that don't take new partitions or reassigned partitions for topics.
    */
-  Set<Integer> getBlackListedBrokers(String zkUrl);
+  Set<Integer> getBlackListedBrokers(AdminClient adminClient);
 
 }