Use createTopicIfNotExist instead of AdminClient (#248)

Use createTopicIfNotExist instead of AdminClient
connects the _topicFactory interface’s createTopicIfNotExist.
🙏🏻
Signed-off-by: Andrew Choi <[email protected]>

Author: Andrew Choi

src/main/java/com/linkedin/kmf/common/Utils.java

@@ -31,6 +31,7 @@
 import org.apache.avro.io.Encoder;
 import org.apache.avro.io.JsonEncoder;
 import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.CreateTopicsResult;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.common.errors.TopicExistsException;
 import org.json.JSONObject;
@@ -98,7 +99,12 @@ public static int createTopicIfNotExists(String topic, short replicationFactor,
 
         List<NewTopic> topics = new ArrayList<>();
         topics.add(newTopic);
-        adminClient.createTopics(topics);
+        CreateTopicsResult result = adminClient.createTopics(topics);
+
+        // waits for this topic creation future to complete, and then returns its result.
+        result.values().get(topic).get();
+        LOG.info("CreateTopicsResult: {}.", result.values());
+
       } catch (TopicExistsException e) {
         /* There is a race condition with the consumer. */
         LOG.debug("Monitoring topic " + topic + " already exists in the cluster.", e);

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

@@ -36,7 +36,6 @@
 import kafka.zk.KafkaZkClient;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AdminClientConfig;
-import org.apache.kafka.clients.admin.CreateTopicsResult;
 import org.apache.kafka.clients.admin.ElectLeadersOptions;
 import org.apache.kafka.clients.admin.ElectLeadersResult;
 import org.apache.kafka.clients.admin.NewPartitions;
@@ -230,7 +229,6 @@ static class TopicManagementHelper {
     private final int _replicationFactor;
     private final double _minPartitionsToBrokersRatio;
     private final int _minPartitionNum;
-    private final TopicFactory _topicFactory;
     private final Properties _topicProperties;
     private boolean _preferredLeaderElectionRequested;
     private final int _requestTimeoutMs;
@@ -240,6 +238,7 @@ static class TopicManagementHelper {
     boolean _topicCreationEnabled;
     AdminClient _adminClient;
     String _topic;
+    TopicFactory _topicFactory;
 
 
     @SuppressWarnings("unchecked")
@@ -277,11 +276,8 @@ void maybeCreateTopic() throws Exception {
         int numPartitions = Math.max((int) Math.ceil(brokerCount * _minPartitionsToBrokersRatio), minPartitionNum());
         NewTopic newTopic = new NewTopic(_topic, numPartitions, (short) _replicationFactor);
         newTopic.configs((Map) _topicProperties);
-        CreateTopicsResult createTopicsResult = _adminClient.createTopics(Collections.singletonList(newTopic));
-
-        // waits for this topic creation future to complete, and then returns its result.
-        createTopicsResult.values().get(_topic).get();
-        LOGGER.info("CreateTopicsResult: {}.", createTopicsResult.values());
+        _topicFactory.createTopicIfNotExist(_topic, (short) _replicationFactor, _minPartitionsToBrokersRatio,
+            _topicProperties, _adminClient);
       }
     }
 

src/test/java/com/linkedin/kmf/services/MultiClusterTopicManagementServiceTest.java

@@ -10,6 +10,7 @@
 
 package com.linkedin.kmf.services;
 
+import com.linkedin.kmf.topicfactory.TopicFactory;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Map;
@@ -59,6 +60,7 @@ private void startTest() {
     _topicManagementHelper = Mockito.mock(MultiClusterTopicManagementService.TopicManagementHelper.class);
     _topicManagementHelper._topic = SERVICE_TEST_TOPIC;
     _topicManagementHelper._adminClient = Mockito.mock(AdminClient.class);
+    _topicManagementHelper._topicFactory = Mockito.mock(TopicFactory.class);
     _topicManagementHelper._topicCreationEnabled = true;
   }
 
@@ -94,6 +96,7 @@ protected void MultiClusterTopicManagementServiceTopicCreationTest() throws Exce
        */
       @Override
       public Void answer(InvocationOnMock invocation) throws Throwable {
+
         Mockito.when(_topicManagementHelper._adminClient.describeTopics(Collections.singleton(SERVICE_TEST_TOPIC)))
             .thenReturn(Mockito.mock(DescribeTopicsResult.class));
         Mockito.when(
@@ -115,10 +118,8 @@ public Void answer(InvocationOnMock invocation) throws Throwable {
       }
     };
 
-    Mockito.when(_topicManagementHelper._adminClient.createTopics(Mockito.anyCollection())
-        .values()
-        .get(SERVICE_TEST_TOPIC)
-        .get()).thenAnswer(createKafkaTopicFutureAnswer);
+    Mockito.when(_topicManagementHelper._topicFactory.createTopicIfNotExist(Mockito.anyString(), Mockito.anyShort(),
+        Mockito.anyDouble(), Mockito.any(), Mockito.any())).thenAnswer(createKafkaTopicFutureAnswer);
 
     _topicManagementHelper.maybeCreateTopic();