Measurement of Commit Latency Service Metrics (#225)

Measurement of Commit Latency Service Metrics
1 - Adds setCommitStartTimeMs() functionality
2 - Adds public void recordCommitComplete() for finishing the recording of consumer offset commit.
3 - Implements public void setCommitStartTimeMs(long time) for setting in milliseconds the start time of consumer offset commit.
4 - Implements public long commitStartTimeMs() for retrieving the start time of consumer offset commit.

Author: Andrew Choi

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

@@ -30,25 +30,76 @@ public class CommitLatencyMetrics {
   private static final String METRIC_GROUP_NAME = "commit-latency-service";
   private static final Logger LOG = LoggerFactory.getLogger(CommitLatencyMetrics.class);
   private final Sensor _commitOffsetLatency;
+  private long _commitStartTimeMs;
+  private volatile boolean _inProgressCommit;
 
   /**
    * Metrics for Calculating the offset commit latency of a consumer.
    * @param metrics the commit offset metrics
    * @param tags the tags associated, i.e) kmf.services:name=single-cluster-monitor
    */
   CommitLatencyMetrics(Metrics metrics, Map<String, String> tags, int latencyPercentileMaxMs, int latencyPercentileGranularityMs) {
+    _inProgressCommit = false;
     _commitOffsetLatency = metrics.sensor("commit-offset-latency");
     _commitOffsetLatency.add(new MetricName("commit-offset-latency-ms-avg", METRIC_GROUP_NAME, "The average latency in ms of committing offset", tags), new Avg());
     _commitOffsetLatency.add(new MetricName("commit-offset-latency-ms-max", METRIC_GROUP_NAME, "The maximum latency in ms of committing offset", tags), new Max());
 
+    if (latencyPercentileGranularityMs == 0) {
+      throw new IllegalArgumentException("The latency percentile granularity was incorrectly passed a zero value.");
+    }
+
+    // 2 extra buckets exist which are respectively designated for values which are less than 0.0 or larger than max.
     int bucketNum = latencyPercentileMaxMs / latencyPercentileGranularityMs + 2;
     int sizeInBytes = bucketNum * 4;
     _commitOffsetLatency.add(new Percentiles(sizeInBytes, latencyPercentileMaxMs, Percentiles.BucketSizing.CONSTANT,
         new Percentile(new MetricName("commit-offset-latency-ms-99th", METRIC_GROUP_NAME, "The 99th percentile latency of committing offset", tags), 99.0),
         new Percentile(new MetricName("commit-offset-latency-ms-999th", METRIC_GROUP_NAME, "The 99.9th percentile latency of committing offset", tags), 99.9),
         new Percentile(new MetricName("commit-offset-latency-ms-9999th", METRIC_GROUP_NAME, "The 99.99th percentile latency of committing offset", tags), 99.99)));
-
     LOG.info("{} was constructed successfully.", this.getClass().getSimpleName());
+  }
 
+  /**
+   * start the recording of consumer offset commit
+   * @throws Exception if the offset commit is already in progress.
+   */
+  public void recordCommitStart() throws Exception {
+    if (!_inProgressCommit) {
+      this.setCommitStartTimeMs(System.currentTimeMillis());
+      _inProgressCommit = true;
+    } else {
+      // inProgressCommit is already set to TRUE;
+      throw new Exception("Offset commit is already in progress.");
+    }
+  }
+
+  /**
+   * finish the recording of consumer offset commit
+   */
+  public void recordCommitComplete() {
+    if (_inProgressCommit) {
+      long commitCompletedMs = System.currentTimeMillis();
+      long commitStartMs = this.commitStartTimeMs();
+      this._commitOffsetLatency.record(commitCompletedMs - commitStartMs);
+      _inProgressCommit = false;
+    } else {
+      // inProgressCommit is already set to FALSE;
+      LOG.error("Offset commit is not in progress. CommitLatencyMetrics shouldn't completing a record commit here.");
+    }
+  }
+
+  /**
+   * set in milliseconds the start time of consumer offset commit
+   * @param time commit start time in ms
+   */
+  public void setCommitStartTimeMs(long time) {
+    _commitStartTimeMs = time;
+  }
+
+  /**
+   * retrieve the start time of consumer offset commit
+   * @return _commitStartTimeMs
+   */
+  public long commitStartTimeMs() {
+    return _commitStartTimeMs;
   }
 }

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

@@ -50,11 +50,12 @@ public class ConsumeService implements Service {
   private static Metrics metrics;
   private final AtomicBoolean _running;
   private final KMBaseConsumer _baseConsumer;
-  private int _latencySlaMs;
+  private final int _latencySlaMs;
   private ConsumeMetrics _sensors;
   private Thread _consumeThread;
-  private AdminClient _adminClient;
+  private final AdminClient _adminClient;
   private CommitAvailabilityMetrics _commitAvailabilityMetrics;
+  private CommitLatencyMetrics _commitLatencyMetrics;
   private String _topic;
   private String _name;
   private static final String METRIC_GROUP_NAME = "consume-service";
@@ -79,6 +80,7 @@ public ConsumeService(String name,
       tags.put(TAGS_NAME, name);
       _topic = consumerFactory.topic();
       _sensors = new ConsumeMetrics(metrics, tags, consumerFactory.latencyPercentileMaxMs(), consumerFactory.latencyPercentileGranularityMs());
+      _commitLatencyMetrics = new CommitLatencyMetrics(metrics, tags, consumerFactory.latencyPercentileMaxMs(), consumerFactory.latencyPercentileGranularityMs());
       _commitAvailabilityMetrics = new CommitAvailabilityMetrics(metrics, tags);
       _consumeThread = new Thread(() -> {
         try {
@@ -125,7 +127,6 @@ record = _baseConsumer.receive();
         continue;
       }
       int partition = record.partition();
-
       /* Commit availability and commit latency service */
       try {
         /* Call commitAsync, wait for a NON-NULL return value (see https://issues.apache.org/jira/browse/KAFKA-6183) */
@@ -137,24 +138,24 @@ public void onComplete(Map<TopicPartition, OffsetAndMetadata> topicPartitionOffs
               _commitAvailabilityMetrics._failedCommitOffsets.record();
             } else {
               _commitAvailabilityMetrics._offsetsCommitted.record();
+              _commitLatencyMetrics.recordCommitComplete();
             }
           }
         };
 
         /* Current timestamp to perform subtraction*/
         long currTimeMillis = System.currentTimeMillis();
 
-        /* 5 seconds consumer offset commit interval. */
+        /* 4 seconds consumer offset commit interval. */
         long timeDiffMillis = TimeUnit.SECONDS.toMillis(COMMIT_TIME_INTERVAL);
 
         if (currTimeMillis - _baseConsumer.lastCommitted() >= timeDiffMillis) {
           /* commit the consumer offset asynchronously with a callback. */
           _baseConsumer.commitAsync(commitCallback);
-
+          _commitLatencyMetrics.recordCommitStart();
           /* Record the current time for the committed consumer offset */
           _baseConsumer.updateLastCommit();
         }
-
       } catch (Exception exception) {
         LOG.error("Exception while trying to perform an asynchronous commit.", exception);
         _commitAvailabilityMetrics._failedCommitOffsets.record();

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

@@ -51,13 +51,17 @@ public GraphiteMetricsReporterService(Map<String, Object> props, String name)
 
   @Override
   public synchronized void start() {
-    _executor.scheduleAtFixedRate(() -> {
-      try {
-        reportMetrics();
-      } catch (Exception e) {
-        LOG.error(_name + "/GraphiteMetricsReporterService failed to report metrics", e);
+    _executor.scheduleAtFixedRate(new Runnable() {
+      @Override
+      public void run() {
+        try {
+          GraphiteMetricsReporterService.this.reportMetrics();
+        } catch (Exception e) {
+          LOG.error(_name + "/GraphiteMetricsReporterService failed to report metrics",
+              e);
+        }
       }
-    }, _reportIntervalSec, _reportIntervalSec, TimeUnit.SECONDS
+      }, _reportIntervalSec, _reportIntervalSec, TimeUnit.SECONDS
     );
     LOG.info("{}/GraphiteMetricsReporterService started", _name);
   }

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

@@ -43,7 +43,7 @@ public class ConsumeServiceTest {
   private static final String TAGS_NAME = "name";
   private static final String METRIC_GROUP_NAME = "commit-availability-service";
   /* thread start delay in seconds */
-  private static final long THREAD_START_DELAY = 4;
+  private static final long THREAD_START_DELAY_SECONDS = 4;
   private static final String TAG_NAME_VALUE = "name";
   private static final long MOCK_LAST_COMMITTED_OFFSET = System.currentTimeMillis();
   private static final int PARTITION = 2;
@@ -83,7 +83,7 @@ public void commitAvailabilityTest() throws Exception {
     Assert.assertTrue(consumeService.isRunning());
 
     /* in milliseconds */
-    long threadStartDelay = 1000 * THREAD_START_DELAY;
+    long threadStartDelay = TimeUnit.SECONDS.toMillis(THREAD_START_DELAY_SECONDS);
 
     /* Thread.sleep safe to do here instead of ScheduledExecutorService
     *  We want to sleep current thread so that consumeService can start running for enough seconds. */
@@ -112,7 +112,7 @@ public void commitLatencyTest() throws Exception {
     Assert.assertTrue(consumeService.isRunning());
 
     /* in milliseconds */
-    long threadStartDelay = TimeUnit.SECONDS.toMillis(THREAD_START_DELAY);
+    long threadStartDelay = TimeUnit.SECONDS.toMillis(THREAD_START_DELAY_SECONDS);
 
     /* Thread.sleep safe to do here instead of ScheduledExecutorService
      *  We want to sleep current thread so that consumeService can start running for enough seconds. */