[WIP] Enable SSL Configurations for ISSUE #178 (#187)

This will be merged after the PR (#181) is merged. and then this PR rebased. Addresses the issue: ISSUE #178 🖌
Signed-off-by: Andrew Choi <[email protected]>

Author: Andrew Choi

src/main/java/com/linkedin/kmf/apps/MultiClusterMonitor.java

@@ -73,8 +73,8 @@ private Map<String, Object> createMultiClusterTopicManagementServiceProps(Map<St
   @Override
   public void start() {
     _multiClusterTopicManagementService.start();
-    CompletableFuture<Void> completableFuture = _multiClusterTopicManagementService.topicManagementReady();
-    completableFuture.thenRun(() -> {
+    CompletableFuture<Void> topicManagementReady = _multiClusterTopicManagementService.topicManagementReady();
+    topicManagementReady.thenRun(() -> {
       _produceService.start();
       _consumeService.start();
     });

src/main/java/com/linkedin/kmf/apps/SingleClusterMonitor.java

@@ -52,7 +52,9 @@ public class SingleClusterMonitor implements App {
   public SingleClusterMonitor(Map<String, Object> props, String name) throws Exception {
     _name = name;
     _topicManagementService = new TopicManagementService(props, name);
+
     CompletableFuture<Void> topicPartitionReady = _topicManagementService.topicPartitionReady();
+
     _produceService = new ProduceService(props, name);
     _consumeService = new ConsumeService(props, name, topicPartitionReady);
   }
@@ -78,16 +80,21 @@ public void stop() {
 
   @Override
   public boolean isRunning() {
+    boolean isRunning = true;
+
     if (!_topicManagementService.isRunning()) {
+      isRunning = false;
       LOG.info("_topicManagementService not running.");
     }
     if (!_produceService.isRunning()) {
+      isRunning = false;
       LOG.info("_produceService not running.");
     }
     if (!_consumeService.isRunning()) {
+      isRunning = false;
       LOG.info("_consumeService not Running.");
     }
-    return _topicManagementService.isRunning() && _produceService.isRunning() && _consumeService.isRunning();
+    return isRunning;
   }
 
   @Override
@@ -279,7 +286,6 @@ public static void main(String[] args) throws Exception {
       props.put(TopicManagementServiceConfig.TOPIC_REPLICATION_FACTOR_CONFIG, res.getInt("replicationFactor"));
     if (res.getInt("rebalanceMs") != null)
       props.put(MultiClusterTopicManagementServiceConfig.REBALANCE_INTERVAL_MS_CONFIG, res.getInt("rebalanceMs"));
-
     SingleClusterMonitor app = new SingleClusterMonitor(props, "single-cluster-monitor");
     app.start();
 

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

@@ -60,7 +60,7 @@ public class ConsumeService implements Service {
   private final String _name;
   private ConsumeMetrics _sensors;
   private final KMBaseConsumer _consumer;
-  private Thread _thread;
+  private Thread _consumeThread;
   private final int _latencyPercentileMaxMs;
   private final int _latencyPercentileGranularityMs;
   private final AtomicBoolean _running;
@@ -104,6 +104,10 @@ public ConsumeService(Map<String, Object> props, String name, CompletableFuture<
 
     // Assign config specified for consumer. This has the highest priority.
     consumerProps.putAll(consumerPropsOverride);
+
+    if (props.containsKey(ConsumeServiceConfig.CONSUMER_PROPS_CONFIG)) {
+      props.forEach(consumerProps::putIfAbsent);
+    }
     _consumer = (KMBaseConsumer) Class.forName(consumerClassName).getConstructor(String.class, Properties.class).newInstance(topic, consumerProps);
     topicPartitionReady.thenRun(() -> {
       MetricConfig metricConfig = new MetricConfig().samples(60).timeWindow(1000, TimeUnit.MILLISECONDS);
@@ -114,16 +118,15 @@ public ConsumeService(Map<String, Object> props, String name, CompletableFuture<
       tags.put("name", _name);
       _adminClient = AdminClient.create(props);
       _sensors = new ConsumeMetrics(metrics, tags, topic, topicPartitionReady);
-      _thread = new Thread(() -> {
+      _consumeThread = new Thread(() -> {
         try {
           consume();
         } catch (Exception e) {
           LOG.error(_name + "/ConsumeService failed", e);
         }
       }, _name + " consume-service");
-      _thread.setDaemon(true);
+      _consumeThread.setDaemon(true);
     });
-
   }
 
   private void consume() throws Exception {
@@ -185,7 +188,7 @@ record = _consumer.receive();
   @Override
   public synchronized void start() {
     if (_running.compareAndSet(false, true)) {
-      _thread.start();
+      _consumeThread.start();
       LOG.info("{}/ConsumeService started.", _name);
     }
   }
@@ -209,7 +212,7 @@ public void awaitShutdown() {
 
   @Override
   public boolean isRunning() {
-    return _running.get() && _thread.isAlive();
+    return _running.get() && _consumeThread.isAlive();
   }
 
   private class ConsumeMetrics {

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

@@ -79,7 +79,9 @@ public class MultiClusterTopicManagementService implements Service {
   private final int _scheduleIntervalMs;
   private final long _preferredLeaderElectionIntervalMs;
   private final ScheduledExecutorService _executor;
-  private final CompletableFuture<Void> _completableFuture;
+
+  private final CompletableFuture<Void> _topicManagementReady;
+
 
   @SuppressWarnings("unchecked")
   public MultiClusterTopicManagementService(Map<String, Object> props, String serviceName) throws Exception {
@@ -91,14 +93,14 @@ public MultiClusterTopicManagementService(Map<String, Object> props, String serv
     _topicManagementByCluster = initializeTopicManagementHelper(propsByCluster, topic);
     _scheduleIntervalMs = config.getInt(MultiClusterTopicManagementServiceConfig.REBALANCE_INTERVAL_MS_CONFIG);
     _preferredLeaderElectionIntervalMs = config.getLong(MultiClusterTopicManagementServiceConfig.PREFERRED_LEADER_ELECTION_CHECK_INTERVAL_MS_CONFIG);
-    _completableFuture = new CompletableFuture<>();
+    _topicManagementReady = new CompletableFuture<>();
     _executor = Executors.newSingleThreadScheduledExecutor(
       r -> new Thread(r, _serviceName + "-multi-cluster-topic-management-service"));
     _topicPartitionReady.complete(null);
   }
 
   public CompletableFuture<Void> topicManagementReady() {
-    return _completableFuture;
+    return _topicManagementReady;
   }
 
   public CompletableFuture<Void> topicPartitionReady() {
@@ -178,7 +180,7 @@ public void run() {
         for (TopicManagementHelper helper : _topicManagementByCluster.values()) {
           helper.maybeAddPartitions(minPartitionNum);
         }
-        _completableFuture.complete(null);
+        _topicManagementReady.complete(null);
         for (Map.Entry<String, TopicManagementHelper> entry : _topicManagementByCluster.entrySet()) {
           String clusterName = entry.getKey();
           TopicManagementHelper helper = entry.getValue();
@@ -261,7 +263,9 @@ static class TopicManagementHelper {
       Map topicFactoryConfig = props.containsKey(TopicManagementServiceConfig.TOPIC_FACTORY_PROPS_CONFIG) ?
           (Map) props.get(TopicManagementServiceConfig.TOPIC_FACTORY_PROPS_CONFIG) : new HashMap();
       _topicFactory = (TopicFactory) Class.forName(topicFactoryClassName).getConstructor(Map.class).newInstance(topicFactoryConfig);
+
       _adminClient = constructAdminClient(props);
+      LOG.info("{} configs: {}", _adminClient.getClass().getSimpleName(), props);
     }
 
     @SuppressWarnings("unchecked")
@@ -277,8 +281,6 @@ void maybeCreateTopic() throws Exception {
     }
 
     AdminClient constructAdminClient(Map<String, Object> props) {
-      props.putIfAbsent(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, _bootstrapServers);
-      props.putIfAbsent(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, _requestTimeoutMs);
       return AdminClient.create(props);
     }
 

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

@@ -113,6 +113,7 @@ public ProduceService(Map<String, Object> props, String name) throws Exception {
         throw new ConfigException("Override must not contain " + property + " config.");
       }
     }
+
     _adminClient = AdminClient.create(props);
 
     if (producerClass.equals(NewProducer.class.getCanonicalName()) || producerClass.equals(NewProducer.class.getSimpleName())) {
@@ -121,7 +122,7 @@ public ProduceService(Map<String, Object> props, String name) throws Exception {
       _producerClassName = producerClass;
     }
 
-    initializeProducer();
+    initializeProducer(props);
 
     _produceExecutor = Executors.newScheduledThreadPool(_threadsNum, new ProduceServiceThreadFactory());
     _handleNewPartitionsExecutor = Executors.newSingleThreadScheduledExecutor(new HandleNewPartitionsThreadFactory());
@@ -135,7 +136,7 @@ public ProduceService(Map<String, Object> props, String name) throws Exception {
     _sensors = new ProduceMetrics(metrics, tags);
   }
 
-  private void initializeProducer() throws Exception {
+  private void initializeProducer(Map<String, Object> props) throws Exception {
     Properties producerProps = new Properties();
     // Assign default config. This has the lowest priority.
     producerProps.put(ProducerConfig.ACKS_CONFIG, "-1");
@@ -151,6 +152,10 @@ private void initializeProducer() throws Exception {
     // Assign config specified for producer. This has the highest priority.
     producerProps.putAll(_producerPropsOverride);
 
+    if (props.containsKey(ProduceServiceConfig.PRODUCER_PROPS_CONFIG)) {
+      props.forEach(producerProps::putIfAbsent);
+    }
+
     _producer = (KMBaseProducer) Class.forName(_producerClassName).getConstructor(Properties.class).newInstance(producerProps);
     LOG.info("{}/ProduceService is initialized.", _name);
   }
@@ -380,7 +385,7 @@ public void run() {
         }
         _producer.close();
         try {
-          initializeProducer();
+          initializeProducer(new HashMap<>());
         } catch (Exception e) {
           LOG.error("Failed to restart producer.", e);
           throw new IllegalStateException(e);

src/main/java/com/linkedin/kmf/services/configs/ConsumeServiceConfig.java

@@ -14,6 +14,7 @@
 import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.common.config.ConfigDef;
 
+
 public class ConsumeServiceConfig extends AbstractConfig {
 
   private static final ConfigDef CONFIG;
@@ -79,7 +80,6 @@ public class ConsumeServiceConfig extends AbstractConfig {
                                     20000,
                                     ConfigDef.Importance.MEDIUM,
                                     LATENCY_SLA_MS_DOC);
-
   }
 
   public ConsumeServiceConfig(Map<?, ?> props) {