KAFKA-15541: Add iterator-duration metrics

streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java

@@ -90,6 +90,7 @@ public class MeteredKeyValueStore<K, V>
     private Sensor prefixScanSensor;
     private Sensor flushSensor;
     private Sensor e2eLatencySensor;
+    protected Sensor iteratorDurationSensor;
     protected InternalProcessorContext context;
     private StreamsMetricsImpl streamsMetrics;
     private TaskId taskId;
@@ -165,6 +166,7 @@ private void registerMetrics() {
         flushSensor = StateStoreMetrics.flushSensor(taskId.toString(), metricsScope, name(), streamsMetrics);
         deleteSensor = StateStoreMetrics.deleteSensor(taskId.toString(), metricsScope, name(), streamsMetrics);
         e2eLatencySensor = StateStoreMetrics.e2ELatencySensor(taskId.toString(), metricsScope, name(), streamsMetrics);
+        iteratorDurationSensor = StateStoreMetrics.iteratorDurationSensor(taskId.toString(), metricsScope, name(), streamsMetrics);
         StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
                 (config, now) -> numOpenIterators.get());
     }
@@ -486,7 +488,9 @@ public void close() {
             try {
                 iter.close();
             } finally {
-                sensor.record(time.nanoseconds() - startNs);
+                final long duration = time.nanoseconds() - startNs;
+                sensor.record(duration);
+                iteratorDurationSensor.record(duration);
                 numOpenIterators.decrementAndGet();
             }
         }
@@ -532,7 +536,9 @@ public void close() {
             try {
                 iter.close();
             } finally {
-                sensor.record(time.nanoseconds() - startNs);
+                final long duration = time.nanoseconds() - startNs;
+                sensor.record(duration);
+                iteratorDurationSensor.record(duration);
                 numOpenIterators.decrementAndGet();
             }
         }

streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredMultiVersionedKeyQueryIterator.java

@@ -18,6 +18,9 @@
 
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
+
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.streams.state.VersionedRecordIterator;
 import org.apache.kafka.streams.state.VersionedRecord;
 
@@ -26,13 +29,21 @@ public class MeteredMultiVersionedKeyQueryIterator<V> implements VersionedRecord
     private final VersionedRecordIterator<byte[]> iterator;
     private final Function<VersionedRecord<byte[]>, VersionedRecord<V>> deserializeValue;
     private final AtomicInteger numOpenIterators;
+    private final Sensor sensor;
+    private final Time time;
+    private final long startNs;
 
     public MeteredMultiVersionedKeyQueryIterator(final VersionedRecordIterator<byte[]> iterator,
+                                                 final Sensor sensor,
+                                                 final Time time,
                                                  final Function<VersionedRecord<byte[]>, VersionedRecord<V>> deserializeValue,
                                                  final AtomicInteger numOpenIterators) {
         this.iterator = iterator;
         this.deserializeValue = deserializeValue;
         this.numOpenIterators = numOpenIterators;
+        this.sensor = sensor;
+        this.time = time;
+        this.startNs = time.nanoseconds();
         numOpenIterators.incrementAndGet();
     }
 
@@ -42,6 +53,7 @@ public void close() {
         try {
             iterator.close();
         } finally {
+            sensor.record(time.nanoseconds() - startNs);
             numOpenIterators.decrementAndGet();
         }
     }

streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStore.java

@@ -68,6 +68,7 @@ public class MeteredSessionStore<K, V>
     private Sensor flushSensor;
     private Sensor removeSensor;
     private Sensor e2eLatencySensor;
+    private Sensor iteratorDurationSensor;
     private InternalProcessorContext<?, ?> context;
     private TaskId taskId;
 
@@ -134,6 +135,7 @@ private void registerMetrics() {
         flushSensor = StateStoreMetrics.flushSensor(taskId.toString(), metricsScope, name(), streamsMetrics);
         removeSensor = StateStoreMetrics.removeSensor(taskId.toString(), metricsScope, name(), streamsMetrics);
         e2eLatencySensor = StateStoreMetrics.e2ELatencySensor(taskId.toString(), metricsScope, name(), streamsMetrics);
+        iteratorDurationSensor = StateStoreMetrics.iteratorDurationSensor(taskId.toString(), metricsScope, name(), streamsMetrics);
         StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
                 (config, now) -> numOpenIterators.get());
     }
@@ -250,6 +252,7 @@ public KeyValueIterator<Windowed<K>, V> fetch(final K key) {
         return new MeteredWindowedKeyValueIterator<>(
             wrapped().fetch(keyBytes(key)),
             fetchSensor,
+            iteratorDurationSensor,
             streamsMetrics,
             serdes::keyFrom,
             serdes::valueFrom,
@@ -263,6 +266,7 @@ public KeyValueIterator<Windowed<K>, V> backwardFetch(final K key) {
         return new MeteredWindowedKeyValueIterator<>(
             wrapped().backwardFetch(keyBytes(key)),
             fetchSensor,
+            iteratorDurationSensor,
             streamsMetrics,
             serdes::keyFrom,
             serdes::valueFrom,
@@ -277,6 +281,7 @@ public KeyValueIterator<Windowed<K>, V> fetch(final K keyFrom,
         return new MeteredWindowedKeyValueIterator<>(
             wrapped().fetch(keyBytes(keyFrom), keyBytes(keyTo)),
             fetchSensor,
+            iteratorDurationSensor,
             streamsMetrics,
             serdes::keyFrom,
             serdes::valueFrom,
@@ -290,6 +295,7 @@ public KeyValueIterator<Windowed<K>, V> backwardFetch(final K keyFrom,
         return new MeteredWindowedKeyValueIterator<>(
             wrapped().backwardFetch(keyBytes(keyFrom), keyBytes(keyTo)),
             fetchSensor,
+            iteratorDurationSensor,
             streamsMetrics,
             serdes::keyFrom,
             serdes::valueFrom,
@@ -310,6 +316,7 @@ public KeyValueIterator<Windowed<K>, V> findSessions(final K key,
                 earliestSessionEndTime,
                 latestSessionStartTime),
             fetchSensor,
+            iteratorDurationSensor,
             streamsMetrics,
             serdes::keyFrom,
             serdes::valueFrom,
@@ -330,6 +337,7 @@ public KeyValueIterator<Windowed<K>, V> backwardFindSessions(final K key,
                 latestSessionStartTime
             ),
             fetchSensor,
+            iteratorDurationSensor,
             streamsMetrics,
             serdes::keyFrom,
             serdes::valueFrom,
@@ -352,6 +360,7 @@ public KeyValueIterator<Windowed<K>, V> findSessions(final K keyFrom,
                 earliestSessionEndTime,
                 latestSessionStartTime),
             fetchSensor,
+            iteratorDurationSensor,
             streamsMetrics,
             serdes::keyFrom,
             serdes::valueFrom,
@@ -365,6 +374,7 @@ public KeyValueIterator<Windowed<K>, V> findSessions(final long earliestSessionE
         return new MeteredWindowedKeyValueIterator<>(
                 wrapped().findSessions(earliestSessionEndTime, latestSessionEndTime),
                 fetchSensor,
+                iteratorDurationSensor,
                 streamsMetrics,
                 serdes::keyFrom,
                 serdes::valueFrom,
@@ -387,6 +397,7 @@ public KeyValueIterator<Windowed<K>, V> backwardFindSessions(final K keyFrom,
                 latestSessionStartTime
             ),
             fetchSensor,
+            iteratorDurationSensor,
             streamsMetrics,
             serdes::keyFrom,
             serdes::valueFrom,
@@ -458,6 +469,7 @@ private <R> QueryResult<R> runRangeQuery(final Query<R> query,
                     new MeteredWindowedKeyValueIterator<>(
                         rawResult.getResult(),
                         fetchSensor,
+                        iteratorDurationSensor,
                         streamsMetrics,
                         serdes::keyFrom,
                         StoreQueryUtils.getDeserializeValue(serdes, wrapped()),

streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStore.java

@@ -350,7 +350,9 @@ public void close() {
             try {
                 iter.close();
             } finally {
-                sensor.record(time.nanoseconds() - startNs);
+                final long duration = time.nanoseconds() - startNs;
+                sensor.record(duration);
+                iteratorDurationSensor.record(duration);
                 numOpenIterators.decrementAndGet();
             }
         }

streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredVersionedKeyValueStore.java

@@ -264,7 +264,13 @@ private <R> QueryResult<R> runMultiVersionedKeyQuery(final Query<R> query, final
             final QueryResult<VersionedRecordIterator<byte[]>> rawResult = wrapped().query(rawKeyQuery, positionBound, config);
             if (rawResult.isSuccess()) {
                 final MeteredMultiVersionedKeyQueryIterator<V> typedResult =
-                        new MeteredMultiVersionedKeyQueryIterator<V>(rawResult.getResult(), StoreQueryUtils.getDeserializeValue(plainValueSerdes), numOpenIterators);
+                        new MeteredMultiVersionedKeyQueryIterator<V>(
+                            rawResult.getResult(),
+                            iteratorDurationSensor,
+                            time,
+                            StoreQueryUtils.getDeserializeValue(plainValueSerdes),
+                            numOpenIterators
+                        );
                 final QueryResult<MeteredMultiVersionedKeyQueryIterator<V>> typedQueryResult =
                         InternalQueryResultUtil.copyAndSubstituteDeserializedResult(rawResult, typedResult);
                 result = (QueryResult<R>) typedQueryResult;

streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredWindowStore.java

@@ -72,6 +72,7 @@ public class MeteredWindowStore<K, V>
     private Sensor fetchSensor;
     private Sensor flushSensor;
     private Sensor e2eLatencySensor;
+    private Sensor iteratorDurationSensor;
     private InternalProcessorContext<?, ?> context;
     private TaskId taskId;
 
@@ -153,6 +154,7 @@ private void registerMetrics() {
         fetchSensor = StateStoreMetrics.fetchSensor(taskId.toString(), metricsScope, name(), streamsMetrics);
         flushSensor = StateStoreMetrics.flushSensor(taskId.toString(), metricsScope, name(), streamsMetrics);
         e2eLatencySensor = StateStoreMetrics.e2ELatencySensor(taskId.toString(), metricsScope, name(), streamsMetrics);
+        iteratorDurationSensor = StateStoreMetrics.iteratorDurationSensor(taskId.toString(), metricsScope, name(), streamsMetrics);
         StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
                 (config, now) -> numOpenIterators.get());
     }
@@ -239,6 +241,7 @@ public WindowStoreIterator<V> fetch(final K key,
         return new MeteredWindowStoreIterator<>(
             wrapped().fetch(keyBytes(key), timeFrom, timeTo),
             fetchSensor,
+            iteratorDurationSensor,
             streamsMetrics,
             serdes::valueFrom,
             time,
@@ -254,6 +257,7 @@ public WindowStoreIterator<V> backwardFetch(final K key,
         return new MeteredWindowStoreIterator<>(
             wrapped().backwardFetch(keyBytes(key), timeFrom, timeTo),
             fetchSensor,
+            iteratorDurationSensor,
             streamsMetrics,
             serdes::valueFrom,
             time,
@@ -273,6 +277,7 @@ public KeyValueIterator<Windowed<K>, V> fetch(final K keyFrom,
                 timeFrom,
                 timeTo),
             fetchSensor,
+            iteratorDurationSensor,
             streamsMetrics,
             serdes::keyFrom,
             serdes::valueFrom,
@@ -292,6 +297,7 @@ public KeyValueIterator<Windowed<K>, V> backwardFetch(final K keyFrom,
                 timeFrom,
                 timeTo),
             fetchSensor,
+            iteratorDurationSensor,
             streamsMetrics,
             serdes::keyFrom,
             serdes::valueFrom,
@@ -305,6 +311,7 @@ public KeyValueIterator<Windowed<K>, V> fetchAll(final long timeFrom,
         return new MeteredWindowedKeyValueIterator<>(
             wrapped().fetchAll(timeFrom, timeTo),
             fetchSensor,
+            iteratorDurationSensor,
             streamsMetrics,
             serdes::keyFrom,
             serdes::valueFrom,
@@ -318,6 +325,7 @@ public KeyValueIterator<Windowed<K>, V> backwardFetchAll(final long timeFrom,
         return new MeteredWindowedKeyValueIterator<>(
             wrapped().backwardFetchAll(timeFrom, timeTo),
             fetchSensor,
+            iteratorDurationSensor,
             streamsMetrics,
             serdes::keyFrom,
             serdes::valueFrom,
@@ -330,6 +338,7 @@ public KeyValueIterator<Windowed<K>, V> all() {
         return new MeteredWindowedKeyValueIterator<>(
             wrapped().all(),
             fetchSensor,
+            iteratorDurationSensor,
             streamsMetrics,
             serdes::keyFrom,
             serdes::valueFrom,
@@ -343,6 +352,7 @@ public KeyValueIterator<Windowed<K>, V> backwardAll() {
         return new MeteredWindowedKeyValueIterator<>(
             wrapped().backwardAll(),
             fetchSensor,
+            iteratorDurationSensor,
             streamsMetrics,
             serdes::keyFrom,
             serdes::valueFrom,
@@ -420,6 +430,7 @@ private <R> QueryResult<R> runRangeQuery(final Query<R> query,
                     new MeteredWindowedKeyValueIterator<>(
                         rawResult.getResult(),
                         fetchSensor,
+                        iteratorDurationSensor,
                         streamsMetrics,
                         serdes::keyFrom,
                         getDeserializeValue(serdes, wrapped()),
@@ -471,6 +482,7 @@ private <R> QueryResult<R> runKeyQuery(final Query<R> query,
                 final MeteredWindowStoreIterator<V> typedResult = new MeteredWindowStoreIterator<>(
                     rawResult.getResult(),
                     fetchSensor,
+                    iteratorDurationSensor,
                     streamsMetrics,
                     getDeserializeValue(serdes, wrapped()),
                     time,

streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredWindowStoreIterator.java

@@ -28,21 +28,24 @@
 class MeteredWindowStoreIterator<V> implements WindowStoreIterator<V> {
 
     private final WindowStoreIterator<byte[]> iter;
-    private final Sensor sensor;
+    private final Sensor operationSensor;
+    private final Sensor iteratorSensor;
     private final StreamsMetrics metrics;
     private final Function<byte[], V> valueFrom;
     private final long startNs;
     private final Time time;
     private final AtomicInteger numOpenIterators;
 
     MeteredWindowStoreIterator(final WindowStoreIterator<byte[]> iter,
-                               final Sensor sensor,
+                               final Sensor operationSensor,
+                               final Sensor iteratorSensor,
                                final StreamsMetrics metrics,
                                final Function<byte[], V> valueFrom,
                                final Time time,
                                final AtomicInteger numOpenIterators) {
         this.iter = iter;
-        this.sensor = sensor;
+        this.operationSensor = operationSensor;
+        this.iteratorSensor = iteratorSensor;
         this.metrics = metrics;
         this.valueFrom = valueFrom;
         this.startNs = time.nanoseconds();
@@ -67,7 +70,9 @@ public void close() {
         try {
             iter.close();
         } finally {
-            sensor.record(time.nanoseconds() - startNs);
+            final long duration = time.nanoseconds() - startNs;
+            operationSensor.record(duration);
+            iteratorSensor.record(duration);
             numOpenIterators.decrementAndGet();
         }
     }

streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredWindowedKeyValueIterator.java

@@ -30,7 +30,8 @@
 class MeteredWindowedKeyValueIterator<K, V> implements KeyValueIterator<Windowed<K>, V> {
 
     private final KeyValueIterator<Windowed<Bytes>, byte[]> iter;
-    private final Sensor sensor;
+    private final Sensor operationSensor;
+    private final Sensor iteratorSensor;
     private final StreamsMetrics metrics;
     private final Function<byte[], K> deserializeKey;
     private final Function<byte[], V> deserializeValue;
@@ -39,14 +40,16 @@ class MeteredWindowedKeyValueIterator<K, V> implements KeyValueIterator<Windowed
     private final AtomicInteger numOpenIterators;
 
     MeteredWindowedKeyValueIterator(final KeyValueIterator<Windowed<Bytes>, byte[]> iter,
-                                    final Sensor sensor,
+                                    final Sensor operationSensor,
+                                    final Sensor iteratorSensor,
                                     final StreamsMetrics metrics,
                                     final Function<byte[], K> deserializeKey,
                                     final Function<byte[], V> deserializeValue,
                                     final Time time,
                                     final AtomicInteger numOpenIterators) {
         this.iter = iter;
-        this.sensor = sensor;
+        this.operationSensor = operationSensor;
+        this.iteratorSensor = iteratorSensor;
         this.metrics = metrics;
         this.deserializeKey = deserializeKey;
         this.deserializeValue = deserializeValue;
@@ -77,7 +80,9 @@ public void close() {
         try {
             iter.close();
         } finally {
-            sensor.record(time.nanoseconds() - startNs);
+            final long duration = time.nanoseconds() - startNs;
+            operationSensor.record(duration);
+            iteratorSensor.record(duration);
             numOpenIterators.decrementAndGet();
         }
     }