KAFKA-20211: Add group coordinator background metrics (apache#21616)

Add the background-queue-time and background-processing-time histogram
metrics.

Add the background-thread-idle-ratio-avg metric. Unlike the event
processor, we cannot measure idle time when using an ExecutorService.
Instead, we measure busy time and invert it when reporting the metric.

All three background metrics are only reported by the group coordinator
and not the share coordinator.

Reviewers: David Jacot <djacot@confluent.io>

Author: squah-confluent

coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorBackgroundThreadPoolExecutor.java

@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.common.runtime;
+
+import org.apache.kafka.common.utils.ThreadUtils;
+import org.apache.kafka.common.utils.Time;
+
+import java.util.Objects;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A {@link ThreadPoolExecutor} that reports metrics.
+ */
+public class CoordinatorBackgroundThreadPoolExecutor extends ThreadPoolExecutor {
+
+    /**
+     * The time.
+     */
+    private final Time time;
+
+    /**
+     * The coordinator runtime metrics.
+     */
+    private final CoordinatorRuntimeMetrics metrics;
+
+    public CoordinatorBackgroundThreadPoolExecutor(
+        String threadPrefix,
+        int numThreads,
+        Time time,
+        CoordinatorRuntimeMetrics metrics
+    ) {
+        super(
+            numThreads,
+            numThreads,
+            0L,
+            TimeUnit.MILLISECONDS,
+            new LinkedBlockingQueue<>(),
+            ThreadUtils.createThreadFactory(Objects.requireNonNull(threadPrefix) + "%d", false)
+        );
+        this.time = Objects.requireNonNull(time);
+        this.metrics = Objects.requireNonNull(metrics);
+    }
+
+    @Override
+    public void execute(Runnable command) {
+        var queuedTimeMs = time.milliseconds();
+
+        super.execute(() -> {
+            var dequeuedTimeMs = time.milliseconds();
+            metrics.recordBackgroundQueueTime(dequeuedTimeMs - queuedTimeMs);
+
+            try {
+                command.run();
+            } finally {
+                long processingTimeMs = time.milliseconds() - dequeuedTimeMs;
+                metrics.recordBackgroundProcessingTime(processingTimeMs);
+                metrics.recordBackgroundThreadBusyTime((double) processingTimeMs / getCorePoolSize());
+            }
+        });
+    }
+}

coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetrics.java

@@ -80,6 +80,26 @@ public interface CoordinatorRuntimeMetrics extends AutoCloseable {
      */
     void recordThreadIdleTime(double idleTimeMs);
 
+    /**
+     * Update the background queue time.
+     *
+     * @param durationMs The queue time.
+     */
+    void recordBackgroundQueueTime(long durationMs);
+
+    /**
+     * Update the background processing time.
+     *
+     * @param durationMs The background processing time.
+     */
+    void recordBackgroundProcessingTime(long durationMs);
+
+    /**
+     * Record the background thread busy time.
+     * @param busyTimeMs The busy time in milliseconds.
+     */
+    void recordBackgroundThreadBusyTime(double busyTimeMs);
+
     /**
      * Register the event queue size gauge.
      *

coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImpl.java

@@ -18,6 +18,7 @@
 
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.metrics.Gauge;
+import org.apache.kafka.common.metrics.MetricConfig;
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.metrics.Sensor;
 import org.apache.kafka.common.metrics.stats.Avg;
@@ -79,6 +80,16 @@ public class CoordinatorRuntimeMetricsImpl implements CoordinatorRuntimeMetrics
      */
     public static final String BATCH_BUFFER_CACHE_DISCARD_COUNT_METRIC_NAME = "batch-buffer-cache-discard-count";
 
+    /**
+     * The background queue time metric name.
+     */
+    public static final String BACKGROUND_QUEUE_TIME_METRIC_NAME = "background-queue-time-ms";
+
+    /**
+     * The background processing time metric name.
+     */
+    public static final String BACKGROUND_PROCESSING_TIME_METRIC_NAME = "background-processing-time-ms";
+
     /**
      * Metric to count the number of partitions in Loading state.
      */
@@ -153,7 +164,22 @@ public class CoordinatorRuntimeMetricsImpl implements CoordinatorRuntimeMetrics
      */
     private final Sensor flushSensor;
 
-    public CoordinatorRuntimeMetricsImpl(Metrics metrics, String metricsGroup) {
+    /**
+     * The background thread busy sensor. Null when background metrics are not enabled.
+     */
+    private final Sensor backgroundThreadBusySensor;
+
+    /**
+     * The background queue time sensor. Null when background metrics are not enabled.
+     */
+    private final Sensor backgroundQueueTimeSensor;
+
+    /**
+     * The background processing time sensor. Null when background metrics are not enabled.
+     */
+    private final Sensor backgroundProcessingTimeSensor;
+
+    public CoordinatorRuntimeMetricsImpl(Metrics metrics, String metricsGroup, boolean enableBackgroundMetrics) {
         this.metrics = Objects.requireNonNull(metrics);
         this.metricsGroup = Objects.requireNonNull(metricsGroup);
 
@@ -265,6 +291,44 @@ public CoordinatorRuntimeMetricsImpl(Metrics metrics, String metricsGroup) {
                 this.metricsGroup,
                 "The flushes per second."),
             new Rate(TimeUnit.SECONDS, new WindowedCount()));
+
+        if (enableBackgroundMetrics) {
+            this.backgroundThreadBusySensor = metrics.sensor(this.metricsGroup + "-BackgroundThreadBusyRatio");
+            this.backgroundThreadBusySensor.add(
+                metrics.metricName(
+                    "background-thread-idle-ratio-avg",
+                    this.metricsGroup,
+                    "The fraction of time the background threads are idle. This is an average across " +
+                        "all coordinator background threads."),
+                new Rate(TimeUnit.MILLISECONDS) {
+                    @Override
+                    public double measure(MetricConfig config, long now) {
+                        return 1.0 - super.measure(config, now);
+                    }
+                });
+
+            KafkaMetricHistogram backgroundQueueTimeHistogram = KafkaMetricHistogram.newLatencyHistogram(
+                suffix -> kafkaMetricName(
+                    BACKGROUND_QUEUE_TIME_METRIC_NAME + "-" + suffix,
+                    "The " + suffix + " background queue time in milliseconds"
+                )
+            );
+            this.backgroundQueueTimeSensor = metrics.sensor(this.metricsGroup + "-BackgroundQueueTime");
+            this.backgroundQueueTimeSensor.add(backgroundQueueTimeHistogram);
+
+            KafkaMetricHistogram backgroundProcessingTimeHistogram = KafkaMetricHistogram.newLatencyHistogram(
+                suffix -> kafkaMetricName(
+                    BACKGROUND_PROCESSING_TIME_METRIC_NAME + "-" + suffix,
+                    "The " + suffix + " background processing time in milliseconds"
+                )
+            );
+            this.backgroundProcessingTimeSensor = metrics.sensor(this.metricsGroup + "-BackgroundProcessingTime");
+            this.backgroundProcessingTimeSensor.add(backgroundProcessingTimeHistogram);
+        } else {
+            this.backgroundThreadBusySensor = null;
+            this.backgroundQueueTimeSensor = null;
+            this.backgroundProcessingTimeSensor = null;
+        }
     }
 
     /**
@@ -298,6 +362,15 @@ public void close() {
         metrics.removeSensor(eventPurgatoryTimeSensor.name());
         metrics.removeSensor(lingerTimeSensor.name());
         metrics.removeSensor(flushSensor.name());
+        if (backgroundThreadBusySensor != null) {
+            metrics.removeSensor(backgroundThreadBusySensor.name());
+        }
+        if (backgroundQueueTimeSensor != null) {
+            metrics.removeSensor(backgroundQueueTimeSensor.name());
+        }
+        if (backgroundProcessingTimeSensor != null) {
+            metrics.removeSensor(backgroundProcessingTimeSensor.name());
+        }
     }
 
     /**
@@ -372,6 +445,27 @@ public void recordThreadIdleTime(double idleTimeMs) {
         threadIdleSensor.record(idleTimeMs);
     }
 
+    @Override
+    public void recordBackgroundThreadBusyTime(double busyTimeMs) {
+        if (backgroundThreadBusySensor != null) {
+            backgroundThreadBusySensor.record(busyTimeMs);
+        }
+    }
+
+    @Override
+    public void recordBackgroundQueueTime(long durationMs) {
+        if (backgroundQueueTimeSensor != null) {
+            backgroundQueueTimeSensor.record(durationMs);
+        }
+    }
+
+    @Override
+    public void recordBackgroundProcessingTime(long durationMs) {
+        if (backgroundProcessingTimeSensor != null) {
+            backgroundProcessingTimeSensor.record(durationMs);
+        }
+    }
+
     @Override
     public void registerEventQueueSizeGauge(Supplier<Integer> sizeSupplier) {
         metrics.addMetric(eventQueueSize, (Gauge<Long>) (config, now) -> (long) sizeSupplier.get());

coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorBackgroundThreadPoolExecutorTest.java

@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.common.runtime;
+
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.Time;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+public class CoordinatorBackgroundThreadPoolExecutorTest {
+
+    @Test
+    public void testMetrics() throws ExecutionException, InterruptedException, TimeoutException {
+        CoordinatorRuntimeMetrics metrics = mock(CoordinatorRuntimeMetrics.class);
+        Time mockTime = new MockTime();
+        CoordinatorBackgroundThreadPoolExecutor threadPoolExecutor = new CoordinatorBackgroundThreadPoolExecutor(
+            "threaad-pool-",
+            2,
+            mockTime,
+            metrics
+        );
+
+        try {
+            CountDownLatch task1Started = new CountDownLatch(1);
+            CountDownLatch task2Started = new CountDownLatch(1);
+            CountDownLatch task3Started = new CountDownLatch(1);
+            CountDownLatch task1Unblocked = new CountDownLatch(1);
+            CountDownLatch task2Unblocked = new CountDownLatch(1);
+            CountDownLatch task3Unblocked = new CountDownLatch(1);
+
+            Future<?> task1 = threadPoolExecutor.submit(() -> {
+                task1Started.countDown();
+                try {
+                    task1Unblocked.await();
+                } catch (InterruptedException e) { }
+            });
+            Future<?> task2 = threadPoolExecutor.submit(() -> {
+                task2Started.countDown();
+                try {
+                    task2Unblocked.await();
+                } catch (InterruptedException e) { }
+            });
+            Future<?> task3 = threadPoolExecutor.submit(() -> {
+                task3Started.countDown();
+                try {
+                    task3Unblocked.await();
+                } catch (InterruptedException e) { }
+            });
+
+            // Task 1 and task 2 start.
+            task1Started.await();
+            task2Started.await();
+
+            // Task 1 takes 100 ms.
+            mockTime.sleep(100);
+            task1Unblocked.countDown();
+            task1.get(5, TimeUnit.SECONDS);
+
+            // Task 3 starts.
+            task3Started.await();
+
+            // Task 2 takes 500 ms.
+            mockTime.sleep(400);
+            task2Unblocked.countDown();
+            task2.get(5, TimeUnit.SECONDS);
+
+            // Task 3 takes 500 ms.
+            mockTime.sleep(100);
+            task3Unblocked.countDown();
+            task3.get(5, TimeUnit.SECONDS);
+
+            verify(metrics, times(2)).recordBackgroundQueueTime(0);
+            verify(metrics, times(1)).recordBackgroundQueueTime(100);
+            verify(metrics, times(1)).recordBackgroundProcessingTime(100);
+            verify(metrics, times(2)).recordBackgroundProcessingTime(500);
+            verify(metrics, times(1)).recordBackgroundThreadBusyTime(50.0);
+            verify(metrics, times(2)).recordBackgroundThreadBusyTime(250.0);
+        } finally {
+            threadPoolExecutor.shutdown();
+            threadPoolExecutor.awaitTermination(5, TimeUnit.SECONDS);
+        }
+    }
+}