address feedback - part 5

Author: MahsaSeifikar

metadata/src/main/java/org/apache/kafka/controller/QuorumController.java

@@ -406,7 +406,13 @@ public QuorumController build() throws Exception {
 
             KafkaEventQueue queue = null;
             try {
-                queue = new KafkaEventQueue(time, logContext, threadNamePrefix, EventQueue.VoidEvent.INSTANCE, controllerMetrics::updateIdleTime);
+                queue = new KafkaEventQueue(
+                        time,
+                        logContext,
+                        threadNamePrefix,
+                        EventQueue.VoidEvent.INSTANCE,
+                        Optional.of(controllerMetrics::updateIdleTime)
+                );
 
                 return new QuorumController(
                     nonFatalFaultHandler,

metadata/src/main/java/org/apache/kafka/controller/metrics/QuorumControllerMetrics.java

@@ -17,7 +17,6 @@
 
 package org.apache.kafka.controller.metrics;
 
-import org.apache.kafka.common.metrics.MetricConfig;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.server.metrics.KafkaYammerMetrics;
 import org.apache.kafka.server.metrics.TimeRatio;
@@ -71,7 +70,6 @@ public class QuorumControllerMetrics implements AutoCloseable {
 
     private static final String TIME_SINCE_LAST_HEARTBEAT_RECEIVED_METRIC_NAME = "TimeSinceLastHeartbeatReceivedMs";
     private static final String BROKER_ID_TAG = "broker";
-    private static final MetricConfig METRIC_CONFIG = new MetricConfig();
 
     private final Optional<MetricsRegistry> registry;
     private final Time time;
@@ -168,13 +166,13 @@ public Long value() {
         registry.ifPresent(r -> r.newGauge(AVERAGE_IDLE_RATIO, new Gauge<Double>() {
             @Override
             public Double value() {
-                return avgIdleTimeRatio.measure(METRIC_CONFIG, time.milliseconds());
+                return avgIdleTimeRatio.measure();
             }
         }));
     }
 
     public void updateIdleTime(long idleDurationMs) {
-        avgIdleTimeRatio.record(METRIC_CONFIG, (double) idleDurationMs, time.milliseconds());
+        avgIdleTimeRatio.record((double) idleDurationMs, time.milliseconds());
     }
 
     public void addTimeSinceLastHeartbeatMetric(int brokerId) {

server-common/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java

@@ -289,8 +289,10 @@ private void handleEvents() {
                         }
                     } catch (InterruptedException e) {
 
-                        log.warn("Interrupted while waiting for a {} event. " +
-                                "Shutting down event queue", (awaitNs == Long.MAX_VALUE) ? "new" : "deferred");
+                        log.warn(
+                                "Interrupted while waiting for a {} event. Shutting down event queue",
+                                (awaitNs == Long.MAX_VALUE) ? "new" : "deferred"
+                        );
                         interrupted = true;
                     } finally {
                         idleTimeCallback.accept(Math.max(time.milliseconds() - startIdleMs, 0));
@@ -451,7 +453,7 @@ public KafkaEventQueue(
         LogContext logContext,
         String threadNamePrefix
     ) {
-        this(time, logContext, threadNamePrefix, VoidEvent.INSTANCE, null);
+        this(time, logContext, threadNamePrefix, VoidEvent.INSTANCE, Optional.empty());
     }
 
     public KafkaEventQueue(
@@ -460,15 +462,15 @@ public KafkaEventQueue(
         String threadNamePrefix,
         Event cleanupEvent
     ) {
-        this(time, logContext, threadNamePrefix, cleanupEvent, null);
+        this(time, logContext, threadNamePrefix, cleanupEvent, Optional.empty());
     }
 
     public KafkaEventQueue(
         Time time,
         LogContext logContext,
         String threadNamePrefix,
         Event cleanupEvent,
-        Consumer<Long> idleTimeCallback
+        Optional<Consumer<Long>> idleTimeCallback
     ) {
         this.time = time;
         this.cleanupEvent = Objects.requireNonNull(cleanupEvent);
@@ -479,7 +481,7 @@ public KafkaEventQueue(
             this.eventHandler, false);
         this.shuttingDown = false;
         this.interrupted = false;
-        this.idleTimeCallback = idleTimeCallback != null ? idleTimeCallback : __ -> { };
+        this.idleTimeCallback = idleTimeCallback.orElse(__ -> { });
         this.eventHandlerThread.start();
     }
 

server-common/src/main/java/org/apache/kafka/server/metrics/TimeRatio.java

@@ -46,11 +46,26 @@ public TimeRatio(double defaultRatio) {
 
     @Override
     public double measure(MetricConfig config, long currentTimestampMs) {
+        return measure();
+    }
+
+    @Override
+    public void record(MetricConfig config, double value, long currentTimestampMs) {
+        record(value, currentTimestampMs);
+    }
+
+    /**
+     * Measures the ratio of recorded duration to the interval duration
+     * since the last measurement.
+     *
+     * @return The measured ratio value between 0.0 and 1.0
+     */
+    public double measure() {
         if (lastRecordedTimestampMs < 0) {
             // Return the default value if no recordings have been captured.
             return defaultRatio;
         } else {
-            // We measure the ratio over the
+            // We measure the ratio over the interval
             double intervalDurationMs = Math.max(lastRecordedTimestampMs - intervalStartTimestampMs, 0);
             final double ratio;
             if (intervalDurationMs == 0) {
@@ -61,15 +76,20 @@ public double measure(MetricConfig config, long currentTimestampMs) {
                 ratio = totalRecordedDurationMs / intervalDurationMs;
             }
 
-            // The next interval begins at the
+            // The next interval begins at the last recorded timestamp
             intervalStartTimestampMs = lastRecordedTimestampMs;
             totalRecordedDurationMs = 0;
             return ratio;
         }
     }
 
-    @Override
-    public void record(MetricConfig config, double value, long currentTimestampMs) {
+    /**
+     * Records a duration value at the specified timestamp.
+     *
+     * @param value The duration value to record
+     * @param currentTimestampMs The current timestamp in milliseconds
+     */
+    public void record(double value, long currentTimestampMs) {
         if (intervalStartTimestampMs < 0) {
             // Discard the initial value since the value occurred prior to the interval start
             intervalStartTimestampMs = currentTimestampMs;
@@ -78,5 +98,4 @@ public void record(MetricConfig config, double value, long currentTimestampMs) {
             lastRecordedTimestampMs = currentTimestampMs;
         }
     }
-
 }

server-common/src/test/java/org/apache/kafka/queue/KafkaEventQueueTest.java

@@ -30,6 +30,7 @@
 import org.junit.jupiter.api.Timeout;
 
 import java.util.List;
+import java.util.Optional;
 import java.util.OptionalLong;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
@@ -436,10 +437,10 @@ public void testIdleTimeCallback() throws Exception {
                 logContext,
                 "testIdleTimeCallback",
                 EventQueue.VoidEvent.INSTANCE,
-                idleMs -> {
+                Optional.of(idleMs -> {
                     lastIdleTimeMs.set(idleMs);
                     idleCallCount.incrementAndGet();
-                })) {
+                }))) {
 
             // Test 1: Two events with a wait in between using FutureEvent
             CompletableFuture<String> event1 = new CompletableFuture<>();