address feedback - part 3

Author: MahsaSeifikar

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

@@ -532,7 +532,6 @@ private void handleEventEnd(String name, long startProcessingTimeNs) {
             MICROSECONDS.convert(deltaNs, NANOSECONDS));
         performanceMonitor.observeEvent(name, deltaNs);
         controllerMetrics.updateEventQueueProcessingTime(NANOSECONDS.toMillis(deltaNs));
-        controllerMetrics.updateIdleStartTime();
     }
 
     private Throwable handleEventException(
@@ -550,7 +549,6 @@ private Throwable handleEventException(
         } else {
             deltaUs = OptionalLong.empty();
         }
-        controllerMetrics.updateIdleStartTime();
         EventHandlerExceptionInfo info = EventHandlerExceptionInfo.
                 fromInternal(exception, this::latestController);
         int epoch = curClaimEpoch;
@@ -575,7 +573,6 @@ private Throwable handleEventException(
 
     private long updateEventStartMetricsAndGetTime(OptionalLong eventCreatedTimeNs) {
         long now = time.nanoseconds();
-        controllerMetrics.updateIdleEndTime();
         controllerMetrics.incrementOperationsStarted();
         if (eventCreatedTimeNs.isPresent()) {
             controllerMetrics.updateEventQueueTime(NANOSECONDS.toMillis(now - eventCreatedTimeNs.getAsLong()));
@@ -834,7 +831,6 @@ public void run() throws Exception {
 
             // Remember the latest offset and future if it is not already completed
             if (!future.isDone()) {
-                controllerMetrics.updateIdleStartTime();
                 deferredEventQueue.add(resultAndOffset.offset(), this);
             }
         }
@@ -846,7 +842,6 @@ public void handleException(Throwable exception) {
 
         @Override
         public void complete(Throwable exception) {
-            controllerMetrics.updateIdleEndTime();
             if (exception == null) {
                 handleEventEnd(this.toString(), startProcessingTimeNs.getAsLong());
                 future.complete(resultAndOffset.response());
@@ -1493,6 +1488,7 @@ private QuorumController(
         this.queue = queue;
         this.time = time;
         this.controllerMetrics = controllerMetrics;
+        queue.setIdleTimeCallback(controllerMetrics::updateIdleTime);
         this.snapshotRegistry = new SnapshotRegistry(logContext);
         this.deferredEventQueue = new DeferredEventQueue(logContext);
         this.resourceExists = new ConfigResourceExistenceChecker();

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

@@ -89,7 +89,6 @@ public class QuorumControllerMetrics implements AutoCloseable {
     private final AtomicLong newActiveControllers = new AtomicLong(0);
     private final Map<Integer, Long> brokerContactTimesMs = new ConcurrentHashMap<>();
     private final int sessionTimeoutMs;
-    private final AtomicLong idleStartTime = new AtomicLong(-1);
 
     private Consumer<Long> newHistogram(MetricName name, boolean biased) {
         if (registry.isPresent()) {
@@ -174,16 +173,8 @@ public Double value() {
         }));
     }
 
-    public void updateIdleStartTime() {
-        idleStartTime.compareAndExchange(-1, time.milliseconds());
-    }
-
-    public void updateIdleEndTime() {
-        long startTime = idleStartTime.getAndSet(-1);
-        if (startTime != -1) {
-            long idleDurationMs = Math.max(time.milliseconds() - startTime, 0);
-            avgIdleTimeRatio.record(METRIC_CONFIG, (double) idleDurationMs, time.milliseconds());
-        }
+    public void updateIdleTime(long idleDurationMs) {
+        avgIdleTimeRatio.record(METRIC_CONFIG, (double) idleDurationMs, time.milliseconds());
     }
 
     public void addTimeSinceLastHeartbeatMetric(int brokerId) {

metadata/src/test/java/org/apache/kafka/controller/metrics/QuorumControllerMetricsTest.java

@@ -199,22 +199,13 @@ public void testAvgIdleRatio() {
             // No idle time recorded yet
             assertEquals(1.0, avgIdleRatio.value());
 
-            metrics.updateIdleStartTime();
-            time.sleep(10); //initial record is ignored
-            metrics.updateIdleEndTime();
-            time.sleep(20); // wait 20ms non-idle
-            metrics.updateIdleStartTime();
-            time.sleep(20); // 20ms idle
-            metrics.updateIdleEndTime();
+            metrics.updateIdleTime(10);
+            time.sleep(40);
+            metrics.updateIdleTime(20);
             assertEquals(0.5, avgIdleRatio.value());
 
-            metrics.updateIdleStartTime();
-            time.sleep(0);
-            metrics.updateIdleEndTime();
-            time.sleep(19);
-            metrics.updateIdleStartTime();
-            time.sleep(1);
-            metrics.updateIdleEndTime();
+            time.sleep(20);
+            metrics.updateIdleTime(1);
             assertEquals(0.05, avgIdleRatio.value());
 
         } finally {

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

@@ -31,8 +31,10 @@
 import java.util.OptionalLong;
 import java.util.TreeMap;
 import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.UnaryOperator;
 
@@ -278,21 +280,24 @@ private void handleEvents() {
                         remove(toRun);
                         continue;
                     }
-                    if (awaitNs == Long.MAX_VALUE) {
-                        try {
+
+                    long startIdleNs = time.nanoseconds();
+                    try {
+                        if (awaitNs == Long.MAX_VALUE) {
                             cond.await();
-                        } catch (InterruptedException e) {
-                            log.warn("Interrupted while waiting for a new event. " +
-                                "Shutting down event queue");
-                            interrupted = true;
-                        }
-                    } else {
-                        try {
+                        } else {
                             cond.awaitNanos(awaitNs);
-                        } catch (InterruptedException e) {
-                            log.warn("Interrupted while waiting for a deferred event. " +
-                                "Shutting down event queue");
-                            interrupted = true;
+                        }
+                    } catch (InterruptedException e) {
+
+                        log.warn("Interrupted while waiting for a {} event. " +
+                                "Shutting down event queue", (awaitNs == Long.MAX_VALUE) ? "new" : "deferred");
+                        interrupted = true;
+                    } finally {
+                        if (idleTimeCallback != null) {
+                            long idleNs = Math.max(time.nanoseconds() - startIdleNs, 0);
+                            long idleMs = TimeUnit.NANOSECONDS.toMillis(idleNs);
+                            idleTimeCallback.accept(idleMs);
                         }
                     }
                 } finally {
@@ -440,6 +445,12 @@ int size() {
      */
     private boolean interrupted;
 
+    /**
+     * Optional callback for queue idle time tracking.
+     */
+    private Consumer<Long> idleTimeCallback;
+
+
     public KafkaEventQueue(
         Time time,
         LogContext logContext,
@@ -503,6 +514,10 @@ public void beginShutdown(String source) {
         }
     }
 
+    public void setIdleTimeCallback(Consumer<Long> callback) {
+        this.idleTimeCallback = callback;
+    }
+
     @Override
     public int size() {
         return eventHandler.size();