Get topicPartitionsWithRetry and a minor spelling error in services/MultiClusterTopicManagementService (#243)

1 - minor spelling error in services/MultiClusterTopicManagementService
`satisifies`
2 - Get topicPartitionsWithRetry after Creation of Topic
3 - LOGGER renaming
4 - Mark CircleCI as TODO

Author: Andrew Choi

.circleci/config.yml

@@ -1,8 +1,13 @@
-version: 2.1
-jobs:
-  build:
-    docker: 
-      - image: circleci/node:4.8.2 # the primary container, where your job's commands are run
+version: 2  # use CircleCI 2.0
+
+jobs: # a collection of steps
+  build: # runs not using Workflows must have a `build` job as entry point
+
+    docker: # run the steps with Docker
+      - image: circleci/openjdk:8-jdk-stretch # with this image as the primary container; this is where all `steps` will run.
+
     steps:
-      - checkout # check out the code in the project directory
-      - run: echo "Kafka Monitor CircleCI"
+      # TODO: populate the necessary steps for CircleCI
+      - run: echo "Kafka Monitor CircleCI on GitHub"
+
+      # See https://circleci.com/docs/2.0/deployment-integrations/ for deploy examples

src/main/java/com/linkedin/kmf/services/MultiClusterTopicManagementService.java

@@ -73,7 +73,7 @@
  */
 @SuppressWarnings({"rawtypes", "unchecked"})
 public class MultiClusterTopicManagementService implements Service {
-  private static final Logger LOG = LoggerFactory.getLogger(MultiClusterTopicManagementService.class);
+  private static final Logger LOGGER = LoggerFactory.getLogger(MultiClusterTopicManagementService.class);
   private static final String METRIC_GROUP_NAME = "topic-management-service";
   private final CompletableFuture<Void> _topicPartitionResult = new CompletableFuture<>();
   private final AtomicBoolean _isRunning = new AtomicBoolean(false);
@@ -126,15 +126,15 @@ public synchronized void start() {
       Runnable pleRunnable = new PreferredLeaderElectionRunnable();
       _executor.scheduleWithFixedDelay(pleRunnable, _preferredLeaderElectionIntervalMs, _preferredLeaderElectionIntervalMs,
           TimeUnit.MILLISECONDS);
-      LOG.info("{}/MultiClusterTopicManagementService started.", _serviceName);
+      LOGGER.info("{}/MultiClusterTopicManagementService started.", _serviceName);
     }
   }
 
   @Override
   public synchronized void stop() {
     if (_isRunning.compareAndSet(true, false)) {
       _executor.shutdown();
-      LOG.info("{}/MultiClusterTopicManagementService stopped.", _serviceName);
+      LOGGER.info("{}/MultiClusterTopicManagementService stopped.", _serviceName);
     }
   }
 
@@ -148,9 +148,9 @@ public void awaitShutdown() {
     try {
       _executor.awaitTermination(Integer.MAX_VALUE, TimeUnit.MILLISECONDS);
     } catch (InterruptedException e) {
-      LOG.info("Thread interrupted when waiting for {}/MultiClusterTopicManagementService to shutdown", _serviceName);
+      LOGGER.info("Thread interrupted when waiting for {}/MultiClusterTopicManagementService to shutdown", _serviceName);
     }
-    LOG.info("{}/MultiClusterTopicManagementService shutdown completed", _serviceName);
+    LOGGER.info("{}/MultiClusterTopicManagementService shutdown completed", _serviceName);
   }
 
 
@@ -165,7 +165,7 @@ public void run() {
         }
 
         /*
-         * The partition number of the monitor topics should be the minimum partition number that satisifies the following conditions:
+         * The partition number of the monitor topics should be the minimum partition number that satisfies the following conditions:
          * - partition number of the monitor topics across all monitored clusters should be the same
          * - partitionNum / brokerNum >= user-configured partitionsToBrokersRatio.
          * - partitionNum >= user-configured minPartitionNum
@@ -185,13 +185,13 @@ public void run() {
           try {
             helper.maybeReassignPartitionAndElectLeader();
           } catch (IOException | KafkaException e) {
-            LOG.warn(_serviceName + "/MultiClusterTopicManagementService will retry later in cluster " + clusterName, e);
+            LOGGER.warn(_serviceName + "/MultiClusterTopicManagementService will retry later in cluster " + clusterName, e);
           }
         }
       } catch (Throwable t) {
         // Need to catch throwable because there is scala API that can throw NoSuchMethodError in runtime
         // and such error is not caught by compilation
-        LOG.error(_serviceName + "/MultiClusterTopicManagementService will stop due to error.", t);
+        LOGGER.error(_serviceName + "/MultiClusterTopicManagementService will stop due to error.", t);
         stop();
       }
     }
@@ -211,18 +211,20 @@ public void run() {
           try {
             helper.maybeElectLeader();
           } catch (IOException | KafkaException e) {
-            LOG.warn(_serviceName + "/MultiClusterTopicManagementService will retry later in cluster " + clusterName, e);
+            LOGGER.warn(_serviceName + "/MultiClusterTopicManagementService will retry later in cluster " + clusterName, e);
           }
         }
       } catch (Throwable t) {
         /* Need to catch throwable because there is scala API that can throw NoSuchMethodError in runtime
          and such error is not caught by compilation. */
-        LOG.error(_serviceName + "/MultiClusterTopicManagementService will stop due to error.", t);
+        LOGGER.error(_serviceName
+            + "/MultiClusterTopicManagementService/PreferredLeaderElectionRunnable will stop due to an error.", t);
         stop();
       }
     }
   }
 
+  @SuppressWarnings("FieldCanBeLocal")
   static class TopicManagementHelper {
     private final boolean _topicCreationEnabled;
     private final String _topic;
@@ -233,9 +235,9 @@ static class TopicManagementHelper {
     private final TopicFactory _topicFactory;
     private final Properties _topicProperties;
     private boolean _preferredLeaderElectionRequested;
-    private int _requestTimeoutMs;
-    private List _bootstrapServers;
-    private final AdminClient _adminClient;
+    private final int _requestTimeoutMs;
+    private final List _bootstrapServers;
+    AdminClient _adminClient;
 
 
     @SuppressWarnings("unchecked")
@@ -263,7 +265,7 @@ static class TopicManagementHelper {
       _topicFactory = (TopicFactory) Class.forName(topicFactoryClassName).getConstructor(Map.class).newInstance(topicFactoryConfig);
 
       _adminClient = constructAdminClient(props);
-      LOG.info("{} configs: {}", _adminClient.getClass().getSimpleName(), props);
+      LOGGER.info("{} configs: {}", _adminClient.getClass().getSimpleName(), props);
     }
 
     @SuppressWarnings("unchecked")
@@ -274,7 +276,10 @@ void maybeCreateTopic() throws Exception {
         NewTopic newTopic = new NewTopic(_topic, numPartitions, (short) _replicationFactor);
         newTopic.configs((Map) _topicProperties);
         CreateTopicsResult createTopicsResult = _adminClient.createTopics(Collections.singletonList(newTopic));
-        LOG.info("CreateTopicsResult: {}.", createTopicsResult.values());
+
+        // waits for this topic creation future to complete, and then returns its result.
+        createTopicsResult.values().get(_topic).get();
+        LOGGER.info("CreateTopicsResult: {}.", createTopicsResult.values());
       }
     }
 
@@ -288,13 +293,14 @@ int minPartitionNum() throws InterruptedException, ExecutionException {
     }
 
     void maybeAddPartitions(int minPartitionNum) throws ExecutionException, InterruptedException {
-      Collection<String> topicNames = _adminClient.listTopics().names().get();
-      Map<String, KafkaFuture<TopicDescription>> kafkaFutureMap = _adminClient.describeTopics(topicNames).values();
+      Map<String, KafkaFuture<TopicDescription>> kafkaFutureMap =
+          _adminClient.describeTopics(Collections.singleton(_topic)).values();
       KafkaFuture<TopicDescription> topicDescriptions = kafkaFutureMap.get(_topic);
       List<TopicPartitionInfo> partitions = topicDescriptions.get().partitions();
+
       int partitionNum = partitions.size();
       if (partitionNum < minPartitionNum) {
-        LOG.info("{} will increase partition of the topic {} in the cluster from {}"
+        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);
         List<List<Integer>> replicaAssignment = new ArrayList<>(new ArrayList<>());
@@ -339,13 +345,13 @@ void maybeReassignPartitionAndElectLeader() throws Exception {
         int expectedReplicationFactor = Math.max(currentReplicationFactor, _replicationFactor);
 
         if (_replicationFactor < currentReplicationFactor)
-          LOG.debug(
+          LOGGER.debug(
               "Configured replication factor {} is smaller than the current replication factor {} of the topic {} in cluster.",
               _replicationFactor, currentReplicationFactor, _topic);
 
         if (expectedReplicationFactor > currentReplicationFactor && !zkClient
             .reassignPartitionsInProgress()) {
-          LOG.info(
+          LOGGER.info(
               "MultiClusterTopicManagementService will increase the replication factor of the topic {} in cluster"
                   + "from {} to {}", _topic, currentReplicationFactor, expectedReplicationFactor);
           reassignPartitions(zkClient, brokers, _topic, partitionInfoList.size(),
@@ -362,15 +368,15 @@ void maybeReassignPartitionAndElectLeader() throws Exception {
           expectedProperties.put(key, _topicProperties.get(key));
 
         if (!currentProperties.equals(expectedProperties)) {
-          LOG.info("MultiClusterTopicManagementService will overwrite properties of the topic {} "
+          LOGGER.info("MultiClusterTopicManagementService will overwrite properties of the topic {} "
               + "in cluster from {} to {}.", _topic, currentProperties, expectedProperties);
           zkClient.setOrCreateEntityConfigs(ConfigType.Topic(), _topic, expectedProperties);
         }
 
         if (partitionInfoList.size() >= brokers.size() &&
             someBrokerNotPreferredLeader(partitionInfoList, brokers) && !zkClient
             .reassignPartitionsInProgress()) {
-          LOG.info("{} will reassign partitions of the topic {} in cluster.",
+          LOGGER.info("{} will reassign partitions of the topic {} in cluster.",
               this.getClass().toString(), _topic);
           reassignPartitions(zkClient, brokers, _topic, partitionInfoList.size(),
               expectedReplicationFactor);
@@ -380,7 +386,7 @@ void maybeReassignPartitionAndElectLeader() throws Exception {
         if (partitionInfoList.size() >= brokers.size() &&
             someBrokerNotElectedLeader(partitionInfoList, brokers)) {
           if (!partitionReassigned || !zkClient.reassignPartitionsInProgress()) {
-            LOG.info(
+            LOGGER.info(
                 "MultiClusterTopicManagementService will trigger preferred leader election for the topic {} in "
                     + "cluster.", _topic
             );
@@ -403,7 +409,7 @@ void maybeElectLeader() throws Exception {
         if (!zkClient.reassignPartitionsInProgress()) {
           List<TopicPartitionInfo> partitionInfoList = _adminClient
               .describeTopics(Collections.singleton(_topic)).all().get().get(_topic).partitions();
-          LOG.info(
+          LOGGER.info(
               "MultiClusterTopicManagementService will trigger requested preferred leader election for the"
                   + " topic {} in cluster.", _topic);
           triggerPreferredLeaderElection(partitionInfoList, _topic);
@@ -424,7 +430,7 @@ private void triggerPreferredLeaderElection(List<TopicPartitionInfo> partitionIn
       Set<TopicPartition> topicPartitions = new HashSet<>(partitions);
       ElectLeadersResult electLeadersResult = _adminClient.electLeaders(electionType, topicPartitions, newOptions);
 
-      LOG.info("{}: triggerPreferredLeaderElection - {}", this.getClass().toString(), electLeadersResult.all().get());
+      LOGGER.info("{}: triggerPreferredLeaderElection - {}", this.getClass().toString(), electLeadersResult.all().get());
     }
 
     private static void reassignPartitions(KafkaZkClient zkClient, Collection<Node> brokers, String topic, int partitionCount, int replicationFactor) {
@@ -448,9 +454,9 @@ private static void reassignPartitions(KafkaZkClient zkClient, Collection<Node>
       String currentAssignmentJson = formatAsReassignmentJson(topic, currentAssignment);
       String newAssignmentJson = formatAsReassignmentJson(topic, assignedReplicas);
 
-      LOG.info("Reassign partitions for topic " + topic);
-      LOG.info("Current partition replica assignment " + currentAssignmentJson);
-      LOG.info("New partition replica assignment " + newAssignmentJson);
+      LOGGER.info("Reassign partitions for topic " + topic);
+      LOGGER.info("Current partition replica assignment " + currentAssignmentJson);
+      LOGGER.info("New partition replica assignment " + newAssignmentJson);
       zkClient.createPartitionReassignment(newAssignment);
     }
 
@@ -461,7 +467,7 @@ static int getReplicationFactor(List<TopicPartitionInfo> partitionInfoList) {
       int replicationFactor = partitionInfoList.get(0).replicas().size();
       for (TopicPartitionInfo partitionInfo : partitionInfoList) {
         if (replicationFactor != partitionInfo.replicas().size()) {
-          LOG.warn("Partitions of the topic have different replication factor.");
+          LOGGER.warn("Partitions of the topic have different replication factor.");
           return -1;
         }
       }