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

@@ -288,9 +288,10 @@ private void handleEvents() {
                             cond.awaitNanos(awaitNs);
                         }
                     } 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 +452,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 +461,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 +480,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;
@@ -429,18 +430,16 @@ public void testInterruptedWithDeferredEvents() throws Exception {
     public void testIdleTimeCallback() throws Exception {
         MockTime time = new MockTime();
         AtomicLong lastIdleTimeMs = new AtomicLong(0);
-        AtomicInteger idleCallCount = new AtomicInteger(0);
 
         try (KafkaEventQueue queue = new KafkaEventQueue(
                 time,
                 logContext,
                 "testIdleTimeCallback",
                 EventQueue.VoidEvent.INSTANCE,
-                idleMs -> {
-                    lastIdleTimeMs.set(idleMs);
-                    idleCallCount.incrementAndGet();
-                })) {
-            
+                Optional.of(lastIdleTimeMs::set))) {
+            time.sleep(2);
+            assertEquals(0, lastIdleTimeMs.get(), "Last idle time should be 0ms");
+
             // Test 1: Two events with a wait in between using FutureEvent
             CompletableFuture<String> event1 = new CompletableFuture<>();
             queue.append(new FutureEvent<>(event1, () -> {
@@ -451,19 +450,14 @@ public void testIdleTimeCallback() throws Exception {
 
             long waitTime5Ms = 5;
             time.sleep(waitTime5Ms);
-            
             CompletableFuture<String> event2 = new CompletableFuture<>();
             queue.append(new FutureEvent<>(event2, () -> {
                 time.sleep(1);
                 return "event2-processed";
             }));
             assertEquals("event2-processed", event2.get());
-
-            TestUtils.waitForCondition(
-                    () -> idleCallCount.get() == 2,
-                    "Idle callback should have been called twice"
-            );
-            assertEquals(waitTime5Ms, lastIdleTimeMs.get(), "Last idle time should be 5ms");
+            assertTrue(lastIdleTimeMs.get() >= waitTime5Ms,
+                "Idle time should be at least " + waitTime5Ms + "ms, was: " + lastIdleTimeMs.get());
 
             // Test 2: Deferred event
             long waitTime2Ms = 2;
@@ -473,12 +467,8 @@ public void testIdleTimeCallback() throws Exception {
                     () -> deferredEvent2.complete(null));
             time.sleep(waitTime2Ms);
             deferredEvent2.get();
-
-            TestUtils.waitForCondition(
-                    () -> idleCallCount.get() == 3,
-                    "Idle callback should have been called three times"
-            );
-            assertEquals(3, idleCallCount.get(), "Idle callback should have been called three times");
+            assertTrue(lastIdleTimeMs.get() >= waitTime2Ms,
+                "Idle time should be at least " + waitTime2Ms + "ms, was: " + lastIdleTimeMs.get());
         }
     }
 }