Address feedback

Author: MahsaSeifikar

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

@@ -575,6 +575,7 @@ 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()));
@@ -598,7 +599,6 @@ class ControllerEvent implements EventQueue.Event {
 
         @Override
         public void run() throws Exception {
-            controllerMetrics.updateIdleEndTime();
             startProcessingTimeNs = OptionalLong.of(
                 updateEventStartMetricsAndGetTime(OptionalLong.of(eventCreatedTimeNs)));
             log.debug("Executing {}.", this);
@@ -650,7 +650,6 @@ CompletableFuture<T> future() {
 
         @Override
         public void run() throws Exception {
-            controllerMetrics.updateIdleEndTime();
             startProcessingTimeNs = OptionalLong.of(
                 updateEventStartMetricsAndGetTime(OptionalLong.of(eventCreatedTimeNs)));
             T value = handler.get();
@@ -766,7 +765,6 @@ CompletableFuture<T> future() {
 
         @Override
         public void run() throws Exception {
-            controllerMetrics.updateIdleEndTime();
             // Deferred events set the DOES_NOT_UPDATE_QUEUE_TIME flag to prevent incorrectly
             // including their deferral time in the event queue time.
             startProcessingTimeNs = OptionalLong.of(

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

@@ -17,7 +17,9 @@
 
 package org.apache.kafka.controller.metrics;
 
+import org.apache.kafka.common.metrics.MetricConfig;
 import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.raft.internals.TimeRatio;
 import org.apache.kafka.server.metrics.KafkaYammerMetrics;
 
 import com.yammer.metrics.core.Gauge;
@@ -70,6 +72,7 @@ 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;
@@ -79,7 +82,7 @@ public class QuorumControllerMetrics implements AutoCloseable {
     private final AtomicLong lastAppliedRecordTimestamp = new AtomicLong(0);
     private final Consumer<Long> eventQueueTimeUpdater;
     private final Consumer<Long> eventQueueProcessingTimeUpdater;
-    public final YammerTimeRatio avgIdleTimeRatio;
+    public final TimeRatio avgIdleTimeRatio;
 
     private final AtomicLong timedOutHeartbeats = new AtomicLong(0);
     private final AtomicLong operationsStarted = new AtomicLong(0);
@@ -115,7 +118,7 @@ public Integer value() {
         this.eventQueueTimeUpdater = newHistogram(EVENT_QUEUE_TIME_MS, true);
         this.eventQueueProcessingTimeUpdater = newHistogram(EVENT_QUEUE_PROCESSING_TIME_MS, true);
         this.sessionTimeoutMs = sessionTimeoutMs;
-        this.avgIdleTimeRatio = new YammerTimeRatio(1);
+        this.avgIdleTimeRatio = new TimeRatio(1);
         registry.ifPresent(r -> r.newGauge(LAST_APPLIED_RECORD_OFFSET, new Gauge<Long>() {
             @Override
             public Long value() {
@@ -164,7 +167,12 @@ public Long value() {
                 return newActiveControllers();
             }
         }));
-        registry.ifPresent(r -> r.newGauge(AVERAGE_IDLE_RATIO, this.avgIdleTimeRatio));
+        registry.ifPresent(r -> r.newGauge(AVERAGE_IDLE_RATIO, new Gauge<Double>() {
+            @Override
+            public Double value() {
+                return avgIdleTimeRatio.measure(METRIC_CONFIG, time.milliseconds());
+            }
+        }));
     }
 
     public void updateIdleStartTime() {
@@ -176,7 +184,7 @@ public void updateIdleStartTime() {
     public void updateIdleEndTime() {
         if (this.idleStartTime.isPresent()) {
             long idleDurationMs = Math.max(time.milliseconds() - idleStartTime.getAsLong(), 0);
-            avgIdleTimeRatio.record((double) idleDurationMs, time.milliseconds());
+            avgIdleTimeRatio.record(METRIC_CONFIG, (double) idleDurationMs, time.milliseconds());
             idleStartTime = OptionalLong.empty();
         }
     }

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

@@ -190,7 +190,7 @@ public void testTimeSinceLastHeartbeatReceivedMs() {
     }
 
     @Test
-    public void testYammerTimeRatioIdleTimeTracking() {
+    public void testAvgIdleRatio() {
         MetricsRegistry registry = new MetricsRegistry();
         MockTime time = new MockTime();
         try (QuorumControllerMetrics metrics = new QuorumControllerMetrics(Optional.of(registry), time, 9000)) {

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

@@ -137,7 +137,7 @@ public KafkaRaftMetrics(Metrics metrics, String metricGrpPrefix) {
                 "The ratio of time the Raft IO thread is idle as opposed to " +
                     "doing work (e.g. handling requests or replicating from the leader)"
             ),
-            new KafkaTimeRatio(1.0)
+            new TimeRatio(1.0)
         );
     }
 

raft/src/main/java/org/apache/kafka/raft/internals/KafkaRaftMetrics.java

@@ -16,17 +16,20 @@
  */
 package org.apache.kafka.raft.internals;
 
+import org.apache.kafka.common.metrics.MeasurableStat;
+import org.apache.kafka.common.metrics.MetricConfig;
+
 /**
- * This metrics agnostic implementation maintain an approximate ratio of
- * the duration of a specific event over all time. For example, this can
- * be used to compute the ratio of time that a thread is busy or idle. The value
- * is approximate since the measurement and recording intervals may not be aligned.
+ * Maintains an approximate ratio of the duration of a specific event
+ * over all time. For example, this can be used to compute the ratio of
+ * time that a thread is busy or idle. The value is approximate since the
+ * measurement and recording intervals may not be aligned.
  *
  * Note that the duration of the event is assumed to be small relative to
  * the interval of measurement.
+ *
  */
-
-public class TimeRatio {
+public class TimeRatio implements MeasurableStat {
     private long intervalStartTimestampMs = -1;
     private long lastRecordedTimestampMs = -1;
     private double totalRecordedDurationMs = 0;
@@ -37,17 +40,12 @@ public TimeRatio(double defaultRatio) {
         if (defaultRatio < 0.0 || defaultRatio > 1.0) {
             throw new IllegalArgumentException("Invalid ratio: value " + defaultRatio + " is not between 0 and 1.");
         }
+
         this.defaultRatio = defaultRatio;
     }
 
-    /**
-     * Measure the ratio of the total recorded duration over the interval duration.
-     * If no recordings have been captured, it returns the default ratio.
-     * After measuring, it resets the recorded duration and starts a new interval.
-     *
-     * @return The ratio of total recorded duration to the interval duration
-     */
-    public double measure() {
+    @Override
+    public double measure(MetricConfig config, long currentTimestampMs) {
         if (lastRecordedTimestampMs < 0) {
             // Return the default value if no recordings have been captured.
             return defaultRatio;
@@ -70,15 +68,8 @@ public double measure() {
         }
     }
 
-    /**
-     * Record the duration of an event at the current timestamp.
-     * If this is the first record, it initializes the interval start timestamp.
-     * Otherwise, it updates the total recorded duration and last recorded timestamp.
-     *
-     * @param value              The duration of the event in milliseconds
-     * @param currentTimestampMs The current time in milliseconds
-     */
-    public void record(double value, long currentTimestampMs) {
+    @Override
+    public void record(MetricConfig config, double value, long currentTimestampMs) {
         if (intervalStartTimestampMs < 0) {
             // Discard the initial value since the value occurred prior to the interval start
             intervalStartTimestampMs = currentTimestampMs;
@@ -87,4 +78,5 @@ public void record(double value, long currentTimestampMs) {
             lastRecordedTimestampMs = currentTimestampMs;
         }
     }
-}
+
+}

raft/src/main/java/org/apache/kafka/raft/internals/KafkaTimeRatio.java

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.raft.internals;
 
+import org.apache.kafka.common.metrics.MetricConfig;
 import org.apache.kafka.common.utils.MockTime;
 
 import org.junit.jupiter.api.Test;
@@ -26,40 +27,42 @@ class TimeRatioTest {
 
     @Test
     public void testRatio() {
+        MetricConfig config = new MetricConfig();
         MockTime time = new MockTime();
         TimeRatio ratio = new TimeRatio(1.0);
 
-        ratio.record(0.0, time.milliseconds());
+        ratio.record(config, 0.0, time.milliseconds());
         time.sleep(10);
-        ratio.record(10, time.milliseconds());
+        ratio.record(config, 10, time.milliseconds());
         time.sleep(10);
-        ratio.record(0, time.milliseconds());
-        assertEquals(0.5, ratio.measure());
+        ratio.record(config, 0, time.milliseconds());
+        assertEquals(0.5, ratio.measure(config, time.milliseconds()));
 
         time.sleep(10);
-        ratio.record(10, time.milliseconds());
+        ratio.record(config, 10, time.milliseconds());
         time.sleep(40);
-        ratio.record(0, time.milliseconds());
-        assertEquals(0.2, ratio.measure());
+        ratio.record(config, 0, time.milliseconds());
+        assertEquals(0.2, ratio.measure(config, time.milliseconds()));
     }
 
     @Test
     public void testRatioMisalignedWindow() {
+        MetricConfig config = new MetricConfig();
         MockTime time = new MockTime();
         TimeRatio ratio = new TimeRatio(1.0);
 
-        ratio.record(0.0, time.milliseconds());
+        ratio.record(config, 0.0, time.milliseconds());
         time.sleep(10);
-        ratio.record(10, time.milliseconds());
+        ratio.record(config, 10, time.milliseconds());
         time.sleep(10);
 
         // No recordings, so the last 10ms are not counted.
-        assertEquals(1.0, ratio.measure());
+        assertEquals(1.0, ratio.measure(config, time.milliseconds()));
 
         // Now the measurement of 5ms arrives. We measure the time since the last
         // recording, so 5ms/10ms = 0.5.
-        ratio.record(5, time.milliseconds());
-        assertEquals(0.5, ratio.measure());
+        ratio.record(config, 5, time.milliseconds());
+        assertEquals(0.5, ratio.measure(config, time.milliseconds()));
     }
 
 }