Commit Availability Service (#196)

**Commit Availability Service.**
1 - Addresses: Broken ConsumeService on latest release #195 #195
2 - Addresses: #204 
3 - Performs Consumer Offset Commit Availability using Metric Sensors. Does periodically, every few x seconds.
4 - In doing this, public interface `ConsumerFactory` is created to build consumer and pass into `consumeService`.

WIP as a follow up PR : Commit `Latency` Service
Signed-off-by: Andrew Choi <[email protected]>

Author: Andrew Choi

build.gradle

@@ -43,6 +43,7 @@ allprojects {
     compile 'com.signalfx.public:signalfx-codahale:0.0.47'
     compile group: 'org.apache.kafka', name: 'kafka_2.12', version: '2.3.1'
     compile group: 'org.apache.kafka', name: 'kafka-clients', version: '2.3.1'
+    testCompile 'org.mockito:mockito-core:2.24.0'
     testCompile 'org.testng:testng:6.8.8'
   }
 

config/kafka-monitor.properties

@@ -38,8 +38,8 @@
 # or ServiceClassName. The key for the test/service in the json map is used as name to
 # identify the test/service in the log or JMX metrics, which is useful if multiple
 # test/service with the same class.name are run in the same Kafka Monitor process.
-# 
-# If using Secure Socket Layer for security protocol, SSL properties must be defined under 
+#
+# If using Secure Socket Layer for security protocol, SSL properties must be defined under
 # produce.producer.props, consume.consumer.props, as well as single-cluster-monitor props
 
 {
@@ -82,21 +82,49 @@
     "class.name": "com.linkedin.kmf.services.DefaultMetricsReporterService",
     "report.interval.sec": 1,
     "report.metrics.list": [
-      "kmf:type=kafka-monitor:offline-runnable-count",
-      "kmf.services:type=produce-service,name=*:produce-availability-avg",
-      "kmf.services:type=consume-service,name=*:consume-availability-avg",
-      "kmf.services:type=produce-service,name=*:records-produced-total",
-      "kmf.services:type=consume-service,name=*:records-consumed-total",
-      "kmf.services:type=consume-service,name=*:records-lost-total",
-      "kmf.services:type=consume-service,name=*:records-lost-rate",
-      "kmf.services:type=consume-service,name=*:records-duplicated-total",
-      "kmf.services:type=consume-service,name=*:records-delay-ms-avg",
-      "kmf.services:type=produce-service,name=*:records-produced-rate",
-      "kmf.services:type=produce-service,name=*:produce-error-rate",
-      "kmf.services:type=consume-service,name=*:consume-error-rate"
+        "kmf:type=kafka-monitor,name=*:offline-runnable-count",
+        "kmf.services:type=produce-service,name=*:produce-availability-avg",
+        "kmf.services:type=consume-service,name=*:consume-availability-avg",
+        "kmf.services:type=commit-availability-service,name=*:offsets-committed-avg",
+        "kmf.services:type=commit-availability-service,name=*:commit-latency-avg",
+        "kmf.services:type=commit-availability-service,name=*:commit-availability-avg",
+        "kmf.services:type=commit-availability-service,name=*:failed-commit-offsets-avg",
+        "kmf.services:type=produce-service,name=*:records-produced-total",
+        "kmf.services:type=consume-service,name=*:records-consumed-total",
+        "kmf.services:type=consume-service,name=*:records-lost-total",
+        "kmf.services:type=consume-service,name=*:records-lost-rate",
+        "kmf.services:type=consume-service,name=*:records-duplicated-total",
+        "kmf.services:type=consume-service,name=*:records-delay-ms-avg",
+        "kmf.services:type=produce-service,name=*:records-produced-rate",
+        "kmf.services:type=produce-service,name=*:produce-error-rate",
+        "kmf.services:type=consume-service,name=*:consume-error-rate",
+        "kmf.services:type=commit-availability-service,name=*:offsets-committed-total",
+        "kmf.services:type=commit-availability-service,name=*:failed-commit-offsets-total"
     ]
   }
 
+#  Example produce-service to produce messages to cluster
+#    "produce-service": {
+#        "class.name": "com.linkedin.kmf.services.ProduceService",
+#        "topic": "kafka-monitor-topic",
+#        "zookeeper.connect": "localhost:2181",
+#        "bootstrap.servers": "localhost:9092",
+#        "consume.latency.sla.ms": "20000",
+#        "consume.consumer.props": {
+#        }
+#    },
+
+#  Example consume-service to consume messages
+#    "consume-service": {
+#        "class.name": "com.linkedin.kmf.services.ConsumeService",
+#        "topic": "kafka-monitor-topic",
+#        "zookeeper.connect": "localhost:2181",
+#        "bootstrap.servers": "localhost:9092",
+#        "consume.latency.sla.ms": "20000",
+#        "consume.consumer.props": {
+#        }
+#    },
+
 #  Example statsd-service to report metrics
 #  "statsd-service": {
 #      "class.name": "com.linkedin.kmf.services.StatsdMetricsReporterService",

src/main/java/com/linkedin/kmf/KafkaMonitor.java

@@ -11,12 +11,15 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.linkedin.kmf.apps.App;
+import com.linkedin.kmf.services.ConsumerFactoryImpl;
 import com.linkedin.kmf.services.Service;
 import java.io.BufferedReader;
 import java.io.FileReader;
+import java.lang.reflect.Constructor;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Executors;
@@ -31,7 +34,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-
 /**
  * This is the main entry point of the monitor.  It reads the configuration and manages the life cycle of the monitoring
  * applications.
@@ -63,13 +65,24 @@ public KafkaMonitor(Map<String, Map> testProps) throws Exception {
         throw new IllegalArgumentException(name + " is not configured with " + CLASS_NAME_CONFIG);
       String className = (String) props.get(CLASS_NAME_CONFIG);
 
-      Class<?> cls = Class.forName(className);
-      if (App.class.isAssignableFrom(cls)) {
+      Class<?> aClass = Class.forName(className);
+      if (App.class.isAssignableFrom(aClass)) {
         App test = (App) Class.forName(className).getConstructor(Map.class, String.class).newInstance(props, name);
         _apps.put(name, test);
-      } else if (Service.class.isAssignableFrom(cls)) {
-        Service service = (Service) Class.forName(className).getConstructor(Map.class, String.class).newInstance(props, name);
-        _services.put(name, service);
+      } else if (Service.class.isAssignableFrom(aClass)) {
+        Constructor<?>[] constructors = Class.forName(className).getConstructors();
+        if (this.constructorContainsFuture(constructors)) {
+          CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+          completableFuture.complete(null);
+          ConsumerFactoryImpl consumerFactory = new ConsumerFactoryImpl(props);
+          Service service = (Service) Class.forName(className)
+              .getConstructor(String.class, CompletableFuture.class, ConsumerFactoryImpl.class)
+              .newInstance(name, completableFuture, consumerFactory);
+          _services.put(name, service);
+        } else {
+          Service service = (Service) Class.forName(className).getConstructor(Map.class, String.class).newInstance(props, name);
+          _services.put(name, service);
+        }
       } else {
         throw new IllegalArgumentException(className + " should implement either " + App.class.getSimpleName() + " or " + Service.class.getSimpleName());
       }
@@ -83,6 +96,15 @@ public KafkaMonitor(Map<String, Map> testProps) throws Exception {
       (config, now) -> _offlineRunnables.size());
   }
 
+  private boolean constructorContainsFuture(Constructor<?>[] constructors) {
+    for (int n = 0; n < constructors[0].getParameterTypes().length; ++n) {
+      if (constructors[0].getParameterTypes()[n].equals(CompletableFuture.class)) {
+        return true;
+      }
+    }
+    return false;
+  }
+
   public synchronized void start() {
     if (!_isRunning.compareAndSet(false, true)) {
       return;
@@ -151,7 +173,6 @@ public static void main(String[] args) throws Exception {
       return;
     }
 
-
     StringBuilder buffer = new StringBuilder();
     try (BufferedReader br = new BufferedReader(new FileReader(args[0].trim()))) {
       String line;

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

@@ -12,6 +12,7 @@
 
 import com.linkedin.kmf.apps.configs.MultiClusterMonitorConfig;
 import com.linkedin.kmf.services.ConsumeService;
+import com.linkedin.kmf.services.ConsumerFactoryImpl;
 import com.linkedin.kmf.services.MultiClusterTopicManagementService;
 import com.linkedin.kmf.services.ProduceService;
 import java.util.HashMap;
@@ -42,7 +43,8 @@ public MultiClusterMonitor(Map<String, Object> props, String name) throws Except
     _multiClusterTopicManagementService = new MultiClusterTopicManagementService(createMultiClusterTopicManagementServiceProps(props, config), name);
     CompletableFuture<Void> topicPartitionReady = _multiClusterTopicManagementService.topicPartitionResult();
     _produceService = new ProduceService(createProduceServiceProps(props, config), name);
-    _consumeService = new ConsumeService(createConsumeServiceProps(props, config), name, topicPartitionReady);
+    ConsumerFactoryImpl consumerFactory = new ConsumerFactoryImpl(createConsumeServiceProps(props, config));
+    _consumeService = new ConsumeService(name, topicPartitionReady, consumerFactory);
   }
 
   @SuppressWarnings("unchecked")

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

@@ -11,16 +11,19 @@
 package com.linkedin.kmf.apps;
 
 import com.linkedin.kmf.services.ConsumeService;
+import com.linkedin.kmf.services.ConsumerFactoryImpl;
 import com.linkedin.kmf.services.DefaultMetricsReporterService;
 import com.linkedin.kmf.services.JettyService;
 import com.linkedin.kmf.services.JolokiaService;
 import com.linkedin.kmf.services.ProduceService;
+import com.linkedin.kmf.services.Service;
 import com.linkedin.kmf.services.TopicManagementService;
 import com.linkedin.kmf.services.configs.ConsumeServiceConfig;
 import com.linkedin.kmf.services.configs.DefaultMetricsReporterServiceConfig;
 import com.linkedin.kmf.services.configs.MultiClusterTopicManagementServiceConfig;
 import com.linkedin.kmf.services.configs.ProduceServiceConfig;
 import com.linkedin.kmf.services.configs.TopicManagementServiceConfig;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
@@ -49,14 +52,20 @@ public class SingleClusterMonitor implements App {
   private final ProduceService _produceService;
   private final ConsumeService _consumeService;
   private final String _name;
+  private final List<Service> _allServices;
 
   public SingleClusterMonitor(Map<String, Object> props, String name) throws Exception {
     _name = name;
     _topicManagementService = new TopicManagementService(props, name);
-
     CompletableFuture<Void> topicPartitionReady = _topicManagementService.topicPartitionResult();
     _produceService = new ProduceService(props, name);
-    _consumeService = new ConsumeService(props, name, topicPartitionReady);
+    ConsumerFactoryImpl consumerFactory = new ConsumerFactoryImpl(props);
+    _consumeService = new ConsumeService(name, topicPartitionReady, consumerFactory);
+    int servicesInitialCapacity = 4;
+    _allServices = new ArrayList<>(servicesInitialCapacity);
+    _allServices.add(_topicManagementService);
+    _allServices.add(_produceService);
+    _allServices.add(_consumeService);
   }
 
   @Override
@@ -72,9 +81,9 @@ public void start() {
 
   @Override
   public void stop() {
-    _topicManagementService.stop();
-    _produceService.stop();
-    _consumeService.stop();
+    for (Service service : _allServices) {
+      service.stop();
+    }
     LOG.info(_name + "/SingleClusterMonitor stopped.");
   }
 
@@ -99,9 +108,9 @@ public boolean isRunning() {
 
   @Override
   public void awaitShutdown() {
-    _topicManagementService.awaitShutdown();
-    _produceService.awaitShutdown();
-    _consumeService.awaitShutdown();
+    for (Service service : _allServices) {
+      service.awaitShutdown();
+    }
   }
 
   /** Get the command-line argument parser. */
@@ -305,7 +314,9 @@ public static void main(String[] args) throws Exception {
       "kmf.services:type=consume-service,name=*:records-delay-ms-avg",
       "kmf.services:type=produce-service,name=*:records-produced-rate",
       "kmf.services:type=produce-service,name=*:produce-error-rate",
-      "kmf.services:type=consume-service,name=*:consume-error-rate"
+      "kmf.services:type=consume-service,name=*:consume-error-rate",
+      "kmf.services:type=consume-service,name=*:commit-latency-avg",
+      "kmf.services:type=consume-service,name=*:commit-availability-avg"
     );
     props.put(DefaultMetricsReporterServiceConfig.REPORT_METRICS_CONFIG, metrics);
 

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

@@ -17,12 +17,15 @@
 import java.lang.reflect.Field;
 import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.common.config.ConfigDef;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 
 /**
  * Generates the table of configuration parameters, their documentation strings and default values.
  */
 public class ConfigDocumentationGenerator {
+  private static final Logger LOG = LoggerFactory.getLogger(ConfigDocumentationGenerator.class);
 
   private static void printHelp() {
     System.out.println("ConfigDocumentationGenerator outputDirectory configClassNames...");

src/main/java/com/linkedin/kmf/consumer/KMBaseConsumer.java

@@ -10,6 +10,12 @@
 
 package com.linkedin.kmf.consumer;
 
+import java.util.Map;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.consumer.OffsetCommitCallback;
+import org.apache.kafka.common.TopicPartition;
+
+
 /**
  * A base consumer used to abstract different consumer classes.
  *
@@ -20,6 +26,18 @@ public interface KMBaseConsumer {
 
   BaseConsumerRecord receive() throws Exception;
 
+  void commitAsync();
+
+  void commitAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback);
+
+  void commitAsync(OffsetCommitCallback callback);
+
+  OffsetAndMetadata committed(TopicPartition tp);
+
   void close();
 
+  long lastCommitted();
+
+  void updateLastCommit();
+
 }

src/main/java/com/linkedin/kmf/consumer/NewConsumer.java

@@ -13,9 +13,15 @@
 import java.time.Duration;
 import java.util.Collections;
 import java.util.Iterator;
+import java.util.Map;
 import java.util.Properties;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.consumer.OffsetCommitCallback;
+import org.apache.kafka.common.TopicPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /*
  * Wrap around the new consumer from Apache Kafka and implement the #KMBaseConsumer interface
@@ -24,6 +30,8 @@ public class NewConsumer implements KMBaseConsumer {
 
   private final KafkaConsumer<String, String> _consumer;
   private Iterator<ConsumerRecord<String, String>> _recordIter;
+  private static final Logger LOG = LoggerFactory.getLogger(NewConsumer.class);
+  private static long lastCommitted;
 
   public NewConsumer(String topic, Properties consumerProperties) {
     _consumer = new KafkaConsumer<>(consumerProperties);
@@ -39,9 +47,39 @@ public BaseConsumerRecord receive() {
     return new BaseConsumerRecord(record.topic(), record.partition(), record.offset(), record.key(), record.value());
   }
 
+  @Override
+  public void commitAsync() {
+    _consumer.commitAsync();
+  }
+
+  @Override
+  public void commitAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback) {
+    _consumer.commitAsync(offsets, callback);
+  }
+
+  @Override
+  public void commitAsync(OffsetCommitCallback callback) {
+    _consumer.commitAsync(callback);
+  }
+
+  @Override
+  public OffsetAndMetadata committed(TopicPartition tp) {
+    return _consumer.committed(tp);
+  }
+
   @Override
   public void close() {
     _consumer.close();
   }
 
+  @Override
+  public long lastCommitted() {
+    return lastCommitted;
+  }
+
+  @Override
+  public void updateLastCommit() {
+    lastCommitted = System.currentTimeMillis();
+  }
+
 }