use lambdas

Author: Maik Toepfer

src/main/java/com/linkedin/xinfra/monitor/services/ConsumeService.java

@@ -152,16 +152,13 @@ record = _baseConsumer.receive();
       int partition = record.partition();
       /* Commit availability and commit latency service */
       /* Call commitAsync, wait for a NON-NULL return value (see https://issues.apache.org/jira/browse/KAFKA-6183) */
-      OffsetCommitCallback commitCallback = new OffsetCommitCallback() {
-        @Override
-        public void onComplete(Map<TopicPartition, OffsetAndMetadata> topicPartitionOffsetAndMetadataMap, Exception kafkaException) {
-          if (kafkaException != null) {
-            LOG.error("Exception while trying to perform an asynchronous commit.", kafkaException);
-            _commitAvailabilityMetrics._failedCommitOffsets.record();
-          } else {
-            _commitAvailabilityMetrics._offsetsCommitted.record();
-            _commitLatencyMetrics.recordCommitComplete();
-          }
+      OffsetCommitCallback commitCallback = (topicPartitionOffsetAndMetadataMap, kafkaException) -> {
+        if (kafkaException != null) {
+          LOG.error("Exception while trying to perform an asynchronous commit.", kafkaException);
+          _commitAvailabilityMetrics._failedCommitOffsets.record();
+        } else {
+          _commitAvailabilityMetrics._offsetsCommitted.record();
+          _commitLatencyMetrics.recordCommitComplete();
         }
       };
 

src/main/java/com/linkedin/xinfra/monitor/services/GraphiteMetricsReporterService.java

@@ -52,17 +52,14 @@ public GraphiteMetricsReporterService(Map<String, Object> props, String name)
 
   @Override
   public synchronized void start() {
-    _executor.scheduleAtFixedRate(new Runnable() {
-      @Override
-      public void run() {
-        try {
-          GraphiteMetricsReporterService.this.reportMetrics();
-        } catch (Exception e) {
-          LOG.error(_name + "/GraphiteMetricsReporterService failed to report metrics",
-              e);
-        }
+    _executor.scheduleAtFixedRate(() -> {
+      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/main/java/com/linkedin/xinfra/monitor/services/OffsetCommitService.java

@@ -134,12 +134,7 @@ public class OffsetCommitService implements Service {
     _consumerNetworkClient = new ConsumerNetworkClient(logContext, kafkaClient, metadata, _time, retryBackoffMs,
         config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG), heartbeatIntervalMs);
 
-    ThreadFactory threadFactory = new ThreadFactory() {
-      @Override
-      public Thread newThread(Runnable runnable) {
-        return new Thread(runnable, serviceName + SERVICE_SUFFIX);
-      }
-    };
+    ThreadFactory threadFactory = runnable -> new Thread(runnable, serviceName + SERVICE_SUFFIX);
     _scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(threadFactory);
 
     LOGGER.info("OffsetCommitService's ConsumerConfig - {}", Utils.prettyPrint(config.values()));

src/main/java/com/linkedin/xinfra/monitor/services/metrics/OffsetCommitServiceMetrics.java

@@ -59,27 +59,24 @@ public OffsetCommitServiceMetrics(final Metrics metrics, final Map<String, Strin
             "The total count of group coordinator unsuccessfully receiving consumer offset commit requests.", tags),
         new CumulativeSum());
 
-    Measurable measurable = new Measurable() {
-      @Override
-      public double measure(MetricConfig config, long now) {
-        double offsetCommitSuccessRate = (double) metrics.metrics()
-            .get(metrics.metricName(SUCCESS_RATE_METRIC, METRIC_GROUP_NAME, tags))
-            .metricValue();
-        double offsetCommitFailureRate = (double) metrics.metrics()
-            .get(metrics.metricName(FAILURE_RATE_METRIC, METRIC_GROUP_NAME, tags))
-            .metricValue();
-
-        if (new Double(offsetCommitSuccessRate).isNaN()) {
-          offsetCommitSuccessRate = 0;
-        }
-
-        if (new Double(offsetCommitFailureRate).isNaN()) {
-          offsetCommitFailureRate = 0;
-        }
-
-        return offsetCommitSuccessRate + offsetCommitFailureRate > 0 ? offsetCommitSuccessRate / (
-            offsetCommitSuccessRate + offsetCommitFailureRate) : 0;
+    Measurable measurable = (config, now) -> {
+      double offsetCommitSuccessRate = (double) metrics.metrics()
+          .get(metrics.metricName(SUCCESS_RATE_METRIC, METRIC_GROUP_NAME, tags))
+          .metricValue();
+      double offsetCommitFailureRate = (double) metrics.metrics()
+          .get(metrics.metricName(FAILURE_RATE_METRIC, METRIC_GROUP_NAME, tags))
+          .metricValue();
+
+      if (new Double(offsetCommitSuccessRate).isNaN()) {
+        offsetCommitSuccessRate = 0;
       }
+
+      if (new Double(offsetCommitFailureRate).isNaN()) {
+        offsetCommitFailureRate = 0;
+      }
+
+      return offsetCommitSuccessRate + offsetCommitFailureRate > 0 ? offsetCommitSuccessRate / (
+          offsetCommitSuccessRate + offsetCommitFailureRate) : 0;
     };
 
     metrics.addMetric(new MetricName("offset-commit-availability-avg", METRIC_GROUP_NAME,

src/test/java/com/linkedin/xinfra/monitor/services/ConsumeServiceTest.java

@@ -144,16 +144,13 @@ private ConsumeService consumeService() throws Exception {
     /* define return value */
     Mockito.when(kmBaseConsumer.lastCommitted()).thenReturn(MOCK_LAST_COMMITTED_OFFSET);
     Mockito.when(kmBaseConsumer.committed(Mockito.any())).thenReturn(new OffsetAndMetadata(FIRST_OFFSET));
-    Mockito.doAnswer(new Answer<Void>() {
-      @Override
-      public Void answer(InvocationOnMock invocationOnMock) {
-        OffsetCommitCallback callback = invocationOnMock.getArgument(0);
-        Map<TopicPartition, OffsetAndMetadata> committedOffsets = new HashMap<>();
-        committedOffsets.put(new TopicPartition(TOPIC, PARTITION), new OffsetAndMetadata(FIRST_OFFSET));
-        callback.onComplete(committedOffsets, null);
+    Mockito.doAnswer((Answer<Void>) invocationOnMock -> {
+      OffsetCommitCallback callback = invocationOnMock.getArgument(0);
+      Map<TopicPartition, OffsetAndMetadata> committedOffsets = new HashMap<>();
+      committedOffsets.put(new TopicPartition(TOPIC, PARTITION), new OffsetAndMetadata(FIRST_OFFSET));
+      callback.onComplete(committedOffsets, null);
 
-        return null;
-      }
+      return null;
     }).when(kmBaseConsumer).commitAsync(Mockito.any(OffsetCommitCallback.class));