diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractMeteredIterator.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractMeteredIterator.java new file mode 100644 index 0000000000000..af815d730f98b --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractMeteredIterator.java @@ -0,0 +1,103 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.streams.state.internals; + +import org.apache.kafka.common.metrics.Sensor; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.streams.state.KeyValueIterator; + +import java.util.Set; +import java.util.concurrent.atomic.LongAdder; + +/** + * Shared metering lifecycle for the metered iterators of the {@code Metered*WithHeaders} stores, + * whatever result type they yield: the {@code KeyValueIterator}s returned by the store's own range/ + * fetch/find methods and the {@code ReadOnlyRecordIterator}s that back the headers-aware IQv2 + * range/window/session query types. + * + *

Every such iterator opens over a raw {@code KeyValueIterator} and needs the + * same bookkeeping: stamp the open time (for the {@code oldest-iterator-open-since-ms} metric), + * register in {@code numOpenIterators}/{@code openIterators}, and on {@link #close()} record the + * operation and iterator-duration sensors and deregister. This base is deliberately result-type + * agnostic -- it implements only {@link MeteredIterator} and does not bind the yielded key/value + * types -- so each subclass declares its own result interface (a {@code KeyValueIterator} or a + * {@code ReadOnlyRecordIterator}) and implements just the parts that genuinely differ: the + * deserializing {@code next()} (and, for the {@code KeyValueIterator}s, a peeking {@code hasNext()} + * and {@code peekNextKey()}). + * + * @param the raw iterator's key type + */ +abstract class AbstractMeteredIterator implements MeteredIterator { + + final KeyValueIterator iter; + private final Sensor operationSensor; + private final Sensor iteratorSensor; + private final Time time; + private final LongAdder numOpenIterators; + private final Set openIterators; + private final long startNs; + private final long startTimestampMs; + + AbstractMeteredIterator(final KeyValueIterator iter, + final Sensor operationSensor, + final Sensor iteratorSensor, + final Time time, + final LongAdder numOpenIterators, + final Set openIterators) { + this.iter = iter; + this.operationSensor = operationSensor; + this.iteratorSensor = iteratorSensor; + this.time = time; + this.numOpenIterators = numOpenIterators; + this.openIterators = openIterators; + this.startNs = time.nanoseconds(); + this.startTimestampMs = time.milliseconds(); + numOpenIterators.increment(); + openIterators.add(this); + } + + // Final: the constructor's openIterators.add(this) sorts through this via the set's + // startTimestamp comparator, i.e. on a not-yet-fully-constructed object. Keeping it final stops a + // subclass from overriding it with something that reads its own not-yet-assigned state. + @Override + public final long startTimestamp() { + return startTimestampMs; + } + + /** + * Delegates to the raw iterator. Subclasses that buffer a peeked element (the + * {@code KeyValueIterator}s) override this to also account for the buffered element. + */ + public boolean hasNext() { + return iter.hasNext(); + } + + // Final: this owns the metering lifecycle (sensor recording, numOpenIterators decrement, + // openIterators deregistration). A subclass that overrode it and forgot super.close() would + // silently drop the decrement and deregistration. Subclasses vary only in next()/hasNext(). + public final void close() { + try { + iter.close(); + } finally { + final long duration = time.nanoseconds() - startNs; + operationSensor.record(duration); + iteratorSensor.record(duration); + numOpenIterators.decrement(); + openIterators.remove(this); + } + } +} diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeaders.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeaders.java index e07051ad1ca3e..f88357dea467e 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeaders.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeaders.java @@ -510,24 +510,13 @@ public KeyValueIterator, AggregationWithHeaders> backwardFindSe } private class MeteredSessionStoreWithHeadersIterator - implements KeyValueIterator, AggregationWithHeaders>, MeteredIterator { + extends AbstractMeteredIterator> + implements KeyValueIterator, AggregationWithHeaders> { - private final KeyValueIterator, byte[]> iter; - private final long startNs; - private final long startTimestampMs; private KeyValue, AggregationWithHeaders> cachedNext; private MeteredSessionStoreWithHeadersIterator(final KeyValueIterator, byte[]> iter) { - this.iter = iter; - this.startNs = time.nanoseconds(); - this.startTimestampMs = time.milliseconds(); - numOpenIterators.increment(); - openIterators.add(this); - } - - @Override - public long startTimestamp() { - return startTimestampMs; + super(iter, fetchSensor, iteratorDurationSensor, time, numOpenIterators, openIterators); } @Override @@ -552,19 +541,6 @@ public KeyValue, AggregationWithHeaders> next() { return KeyValue.pair(windowedKey, value); } - @Override - public void close() { - try { - iter.close(); - } finally { - final long duration = time.nanoseconds() - startNs; - fetchSensor.record(duration); - iteratorDurationSensor.record(duration); - numOpenIterators.decrement(); - openIterators.remove(this); - } - } - @Override public Windowed peekNextKey() { if (cachedNext == null) { @@ -585,30 +561,13 @@ public Windowed peekNextKey() { * non-negative when the window is constructed, so this iterator's {@code next()} can never throw. */ private class MeteredSessionWithHeadersReadOnlyRecordIterator - implements ReadOnlyRecordIterator, AGG>, MeteredIterator { - - private final KeyValueIterator, byte[]> iter; - private final long startNs; - private final long startTimestampMs; + extends AbstractMeteredIterator> + implements ReadOnlyRecordIterator, AGG> { private MeteredSessionWithHeadersReadOnlyRecordIterator( final KeyValueIterator, byte[]> iter ) { - this.iter = iter; - this.startNs = time.nanoseconds(); - this.startTimestampMs = time.milliseconds(); - numOpenIterators.increment(); - openIterators.add(this); - } - - @Override - public long startTimestamp() { - return startTimestampMs; - } - - @Override - public boolean hasNext() { - return iter.hasNext(); + super(iter, fetchSensor, iteratorDurationSensor, time, numOpenIterators, openIterators); } @Override @@ -626,18 +585,5 @@ public ReadOnlyRecord, AGG> next() { ((RecordHeaders) record.headers()).setReadOnly(); return record; } - - @Override - public void close() { - try { - iter.close(); - } finally { - final long duration = time.nanoseconds() - startNs; - fetchSensor.record(duration); - iteratorDurationSensor.record(duration); - numOpenIterators.decrement(); - openIterators.remove(this); - } - } } } diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java index 4aa0b63982da6..9657bef692b1b 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java @@ -657,48 +657,9 @@ private KeyValueIterator> reverseAllInternal( return new MeteredTimestampedKeyValueStoreWithHeadersIterator(store.reverseAll(), allSensor); } - /** - * Shared scaffolding for the metered iterators below: tracks {@code num-open-iterators}, - * {@code oldest-iterator-open-since-ms}, and per-operation iterator duration, and delegates - * closing the wrapped raw iterator. Subclasses only need to implement the deserializing - * {@code next()}/{@code hasNext()} (and, where applicable, {@code peekNextKey()}). - */ - private abstract class AbstractMeteredIterator implements MeteredIterator { - - final KeyValueIterator iter; - private final Sensor sensor; - private final long startNs; - private final long startTimestampMs; - - AbstractMeteredIterator(final KeyValueIterator iter, final Sensor sensor) { - this.iter = iter; - this.sensor = sensor; - this.startNs = time.nanoseconds(); - this.startTimestampMs = time.milliseconds(); - numOpenIterators.increment(); - openIterators.add(this); - } - - @Override - public long startTimestamp() { - return startTimestampMs; - } - - public void close() { - try { - iter.close(); - } finally { - final long duration = time.nanoseconds() - startNs; - sensor.record(duration); - iteratorDurationSensor.record(duration); - numOpenIterators.decrement(); - openIterators.remove(this); - } - } - } - @SuppressWarnings("unchecked") - private class MeteredTimestampedKeyValueStoreWithHeadersQueryIterator extends AbstractMeteredIterator implements KeyValueIterator { + private class MeteredTimestampedKeyValueStoreWithHeadersQueryIterator + extends AbstractMeteredIterator implements KeyValueIterator { private final Function> valueTimestampHeadersDeserializer; @@ -711,7 +672,7 @@ private MeteredTimestampedKeyValueStoreWithHeadersQueryIterator( final Function> valueTimestampHeadersDeserializer, final boolean returnPlainValue ) { - super(iter, sensor); + super(iter, sensor, iteratorDurationSensor, time, numOpenIterators, openIterators); this.valueTimestampHeadersDeserializer = valueTimestampHeadersDeserializer; this.returnPlainValue = returnPlainValue; } @@ -777,7 +738,7 @@ public K peekNextKey() { * {@code next()} throws. */ private class MeteredTimestampedKeyValueStoreWithHeadersReadOnlyRecordIterator - extends AbstractMeteredIterator implements ReadOnlyRecordIterator { + extends AbstractMeteredIterator implements ReadOnlyRecordIterator { private final Function> valueTimestampHeadersDeserializer; @@ -786,15 +747,10 @@ private MeteredTimestampedKeyValueStoreWithHeadersReadOnlyRecordIterator( final Sensor sensor, final Function> valueTimestampHeadersDeserializer ) { - super(iter, sensor); + super(iter, sensor, iteratorDurationSensor, time, numOpenIterators, openIterators); this.valueTimestampHeadersDeserializer = valueTimestampHeadersDeserializer; } - @Override - public boolean hasNext() { - return iter.hasNext(); - } - @Override public ReadOnlyRecord next() { final KeyValue keyValue = iter.next(); @@ -817,7 +773,7 @@ public ReadOnlyRecord next() { } private class MeteredTimestampedKeyValueStoreWithHeadersIterator - extends AbstractMeteredIterator implements KeyValueIterator> { + extends AbstractMeteredIterator implements KeyValueIterator> { private KeyValue> cachedNext; @@ -825,7 +781,7 @@ private MeteredTimestampedKeyValueStoreWithHeadersIterator( final KeyValueIterator iter, final Sensor sensor ) { - super(iter, sensor); + super(iter, sensor, iteratorDurationSensor, time, numOpenIterators, openIterators); } @Override diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedWindowStoreWithHeaders.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedWindowStoreWithHeaders.java index 7fc579160520a..42eaf76e059e8 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedWindowStoreWithHeaders.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedWindowStoreWithHeaders.java @@ -563,25 +563,14 @@ public KeyValueIterator, ValueTimestampHeaders> backwardFetchAll( } private class MeteredTimestampedWindowStoreWithHeadersKeyValueIterator - implements KeyValueIterator, ValueTimestampHeaders>, MeteredIterator { + extends AbstractMeteredIterator> + implements KeyValueIterator, ValueTimestampHeaders> { - private final KeyValueIterator, byte[]> iter; - private final long startNs; - private final long startTimestampMs; private KeyValue, ValueTimestampHeaders> cachedNext; private MeteredTimestampedWindowStoreWithHeadersKeyValueIterator( final KeyValueIterator, byte[]> iter) { - this.iter = iter; - this.startNs = time.nanoseconds(); - this.startTimestampMs = time.milliseconds(); - numOpenIterators.increment(); - openIterators.add(this); - } - - @Override - public long startTimestamp() { - return this.startTimestampMs; + super(iter, fetchSensor, iteratorDurationSensor, time, numOpenIterators, openIterators); } @Override @@ -605,19 +594,6 @@ public KeyValue, ValueTimestampHeaders> next() { return KeyValue.pair(windowedKey, valueTimestampHeaders); } - @Override - public void close() { - try { - iter.close(); - } finally { - final long duration = time.nanoseconds() - startNs; - fetchSensor.record(duration); - iteratorDurationSensor.record(duration); - numOpenIterators.decrement(); - openIterators.remove(this); - } - } - @Override public Windowed peekNextKey() { if (cachedNext == null) { @@ -662,33 +638,17 @@ public Windowed peekNextKey() { * result, or a {@code Windowed} for a range result */ private class MeteredWindowStoreWithHeadersReadOnlyRecordIterator - implements ReadOnlyRecordIterator, V>, MeteredIterator { + extends AbstractMeteredIterator + implements ReadOnlyRecordIterator, V> { - private final KeyValueIterator iter; private final BiFunction> toWindowedKey; - private final long startNs; - private final long startTimestampMs; private MeteredWindowStoreWithHeadersReadOnlyRecordIterator( final KeyValueIterator iter, final BiFunction> toWindowedKey ) { - this.iter = iter; + super(iter, fetchSensor, iteratorDurationSensor, time, numOpenIterators, openIterators); this.toWindowedKey = toWindowedKey; - this.startNs = time.nanoseconds(); - this.startTimestampMs = time.milliseconds(); - numOpenIterators.increment(); - openIterators.add(this); - } - - @Override - public long startTimestamp() { - return startTimestampMs; - } - - @Override - public boolean hasNext() { - return iter.hasNext(); } @Override @@ -722,19 +682,6 @@ public ReadOnlyRecord, V> next() { ((RecordHeaders) record.headers()).setReadOnly(); return record; } - - @Override - public void close() { - try { - iter.close(); - } finally { - final long duration = time.nanoseconds() - startNs; - fetchSensor.record(duration); - iteratorDurationSensor.record(duration); - numOpenIterators.decrement(); - openIterators.remove(this); - } - } } private MeteredWindowedKeyValueIterator meteredWindowedIterator( diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeadersTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeadersTest.java index d57c9c02e2e14..7750027f70c0f 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeadersTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeadersTest.java @@ -67,6 +67,7 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import static org.apache.kafka.common.utils.Utils.mkEntry; @@ -646,25 +647,78 @@ public void shouldTimeIteratorDuration() { setUp(); init(); - final Headers headers = new RecordHeaders(); - headers.add("key1", "value1".getBytes()); - final AggregationWithHeaders valueAndHeaders = AggregationWithHeaders.make(VALUE, headers); + when(innerStore.fetch(KEY_BYTES)) + .thenReturn( + new KeyValueIteratorStub<>(Collections., byte[]>>emptyList().iterator()), + new KeyValueIteratorStub<>(Collections., byte[]>>emptyList().iterator())); + + final KafkaMetric iteratorDurationAvgMetric = metric("iterator-duration-avg"); + final KafkaMetric iteratorDurationMaxMetric = metric("iterator-duration-max"); + assertEquals(Double.NaN, (Double) iteratorDurationAvgMetric.metricValue()); + assertEquals(Double.NaN, (Double) iteratorDurationMaxMetric.metricValue()); + + // Two samples (2ms then 3ms) so avg (2.5ms) and max (3ms) differ -- one sample would leave them + // identical and not actually pin avg. Mirrors the KV sibling shouldTimeIteratorDuration. + try (KeyValueIterator, AggregationWithHeaders> iterator = store.fetch(KEY)) { + mockTime.sleep(2); + } - final AggregationWithHeadersSerializer serializer = new AggregationWithHeadersSerializer<>(Serdes.String().serializer()); - final byte[] serializedValue = serializer.serialize(CHANGELOG_TOPIC, valueAndHeaders); + assertEquals(2.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationAvgMetric.metricValue()); + assertEquals(2.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationMaxMetric.metricValue()); - when(innerStore.fetch(KEY_BYTES)) - .thenReturn(new KeyValueIteratorStub<>( - Collections.singleton(KeyValue.pair(WINDOWED_KEY_BYTES, serializedValue)).iterator())); + try (KeyValueIterator, AggregationWithHeaders> iterator = store.fetch(KEY)) { + mockTime.sleep(3); + } - final KeyValueIterator, AggregationWithHeaders> iterator = store.fetch(KEY); + assertEquals(2.5 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationAvgMetric.metricValue()); + assertEquals(3.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationMaxMetric.metricValue()); + } - mockTime.sleep(100L); + // The above shouldTimeIteratorDuration goes through store.fetch() -> the KeyValueIterator sibling. + // This pins the same close()-path recording for the ReadOnlyRecordIterator that backs + // TimestampedWindowRangeWithHeadersQuery.withKey, whose close() records both the operation sensor + // (fetch) and the iterator-duration sensor via the shared AbstractMeteredIterator lifecycle. + @SuppressWarnings({"unchecked", "rawtypes"}) + @Test + public void shouldTimeIteratorDurationForTimestampedWindowRangeWithHeadersQuery() { + setUp(); + init(); - iterator.close(); + when(innerStore.query(any(), any(PositionBound.class), any(QueryConfig.class))) + .thenReturn( + (QueryResult) QueryResult.forResult(new KeyValueIteratorStub<>( + Collections., byte[]>>emptyList().iterator())), + (QueryResult) QueryResult.forResult(new KeyValueIteratorStub<>( + Collections., byte[]>>emptyList().iterator()))); + + final KafkaMetric iteratorDurationAvgMetric = metric("iterator-duration-avg"); + final KafkaMetric iteratorDurationMaxMetric = metric("iterator-duration-max"); + final KafkaMetric fetchLatencyMetric = metric("fetch-latency-avg"); + assertEquals(Double.NaN, (Double) iteratorDurationAvgMetric.metricValue()); + assertEquals(Double.NaN, (Double) iteratorDurationMaxMetric.metricValue()); + + // Two samples (2ms then 3ms), deterministic under mockTime, so avg (2.5ms) and max (3ms) differ + // and are pinned exactly -- one sample would leave avg == max. + try (ReadOnlyRecordIterator, String> iterator = store.query( + TimestampedWindowRangeWithHeadersQuery.withKey(KEY), + PositionBound.unbounded(), new QueryConfig(false)).getResult()) { + mockTime.sleep(2); + } + + assertEquals(2.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationAvgMetric.metricValue()); + assertEquals(2.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationMaxMetric.metricValue()); + + try (ReadOnlyRecordIterator, String> iterator = store.query( + TimestampedWindowRangeWithHeadersQuery.withKey(KEY), + PositionBound.unbounded(), new QueryConfig(false)).getResult()) { + mockTime.sleep(3); + } - final KafkaMetric iteratorDurationMetric = metric("iterator-duration-avg"); - assertTrue((Double) iteratorDurationMetric.metricValue() > 0.0); + assertEquals(2.5 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationAvgMetric.metricValue()); + assertEquals(3.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationMaxMetric.metricValue()); + // fetchSensor is recorded only from the iterator's close() on this path, so the two samples + // (2ms, 3ms) average to exactly 2.5ms. + assertEquals(2.5 * TimeUnit.MILLISECONDS.toNanos(1), (double) fetchLatencyMetric.metricValue()); } @Test diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeadersTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeadersTest.java index 14087a855ea99..028e3541c0362 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeadersTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeadersTest.java @@ -629,6 +629,52 @@ public void shouldTimeIteratorDuration() { assertEquals(3.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationMaxMetric.metricValue()); } + // The above shouldTimeIteratorDuration goes through metered.all() -> the KeyValueIterator sibling. + // This pins the same close()-path recording for the ReadOnlyRecordIterator that backs + // TimestampedRangeWithHeadersQuery, whose close() records both the operation sensor (get) and the + // iterator-duration sensor. All three Metered*WithHeaders ReadOnlyRecordIterators now share that + // close() via AbstractMeteredIterator, so this also guards the shared lifecycle. + @SuppressWarnings("unchecked") + @Test + public void shouldTimeIteratorDurationForTimestampedRangeWithHeadersQuery() { + setUp(); + when(inner.query(any(), any(PositionBound.class), any(QueryConfig.class))) + .thenReturn( + (QueryResult) QueryResult.forResult(KeyValueIterators.emptyIterator()), + (QueryResult) QueryResult.forResult(KeyValueIterators.emptyIterator())); + init(); + + final KafkaMetric iteratorDurationAvgMetric = metric("iterator-duration-avg"); + final KafkaMetric iteratorDurationMaxMetric = metric("iterator-duration-max"); + final KafkaMetric getLatencyAvgMetric = metric("get-latency-avg"); + assertNotNull(iteratorDurationAvgMetric); + assertNotNull(iteratorDurationMaxMetric); + assertNotNull(getLatencyAvgMetric); + assertEquals(Double.NaN, (Double) iteratorDurationAvgMetric.metricValue()); + assertEquals(Double.NaN, (Double) iteratorDurationMaxMetric.metricValue()); + + // Two samples (2ms then 3ms) so avg (2.5ms) and max (3ms) differ -- one sample would leave them + // identical and not actually pin avg. Mirrors the sibling shouldTimeIteratorDuration above. + try (ReadOnlyRecordIterator iterator = metered.query( + TimestampedRangeWithHeadersQuery.withNoBounds(), PositionBound.unbounded(), new QueryConfig(false)).getResult()) { + mockTime.sleep(2); + } + + assertEquals(2.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationAvgMetric.metricValue()); + assertEquals(2.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationMaxMetric.metricValue()); + + try (ReadOnlyRecordIterator iterator = metered.query( + TimestampedRangeWithHeadersQuery.withNoBounds(), PositionBound.unbounded(), new QueryConfig(false)).getResult()) { + mockTime.sleep(3); + } + + assertEquals(2.5 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationAvgMetric.metricValue()); + assertEquals(3.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationMaxMetric.metricValue()); + // getSensor is recorded only from the iterator's close() on this path, so the two samples + // (2ms, 3ms) average to exactly 2.5ms. + assertEquals(2.5 * TimeUnit.MILLISECONDS.toNanos(1), (double) getLatencyAvgMetric.metricValue()); + } + @SuppressWarnings("unused") @Test public void shouldTrackOldestOpenIteratorTimestamp() { diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedWindowStoreWithHeadersTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedWindowStoreWithHeadersTest.java index 8820319f0e381..c712ab8318bdd 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedWindowStoreWithHeadersTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedWindowStoreWithHeadersTest.java @@ -74,6 +74,7 @@ import java.util.Iterator; import java.util.List; import java.util.Optional; +import java.util.concurrent.TimeUnit; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; @@ -110,6 +111,7 @@ public class MeteredTimestampedWindowStoreWithHeadersTest { private WindowStore innerStoreMock; private final Metrics metrics = new Metrics(new MetricConfig().recordLevel(Sensor.RecordingLevel.DEBUG)); private MeteredTimestampedWindowStoreWithHeaders store; + private MockTime mockTime; private Deserializer keyDeserializer; public void setUp() { @@ -130,11 +132,12 @@ public void setUp() { when(innerStoreMock.name()).thenReturn(STORE_NAME); + mockTime = new MockTime(); store = new MeteredTimestampedWindowStoreWithHeaders<>( innerStoreMock, WINDOW_SIZE_MS, // any size STORE_TYPE, - new MockTime(), + mockTime, Serdes.String(), new ValueTimestampHeadersSerde<>(new SerdeThatDoesntHandleNull()) ); @@ -619,6 +622,84 @@ public void shouldDecrementOpenIteratorsTwiceWhenClosedTwiceForTimestampedWindow assertEquals(-1L, (Long) openIterators.metricValue()); } + // The window store previously had no iterator-duration coverage at all. This mirrors the + // session/KV shouldTimeIteratorDuration: it goes through store.all() -> the KeyValueIterator + // sibling, whose close() records the operation (fetch) and iterator-duration sensors via the + // shared AbstractMeteredIterator lifecycle. + @Test + public void shouldTimeIteratorDuration() { + setUp(); + store.init(context, store); + when(innerStoreMock.all()).thenReturn(windowRangeIterator(List.of()), windowRangeIterator(List.of())); + + final KafkaMetric iteratorDurationAvg = metric("iterator-duration-avg"); + final KafkaMetric iteratorDurationMax = metric("iterator-duration-max"); + assertEquals(Double.NaN, (Double) iteratorDurationAvg.metricValue()); + assertEquals(Double.NaN, (Double) iteratorDurationMax.metricValue()); + + // Two samples (2ms then 3ms) so avg (2.5ms) and max (3ms) differ -- one sample would leave them + // identical and not actually pin avg. + try (KeyValueIterator, ValueTimestampHeaders> iterator = store.all()) { + mockTime.sleep(2); + } + + assertEquals(2.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationAvg.metricValue()); + assertEquals(2.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationMax.metricValue()); + + try (KeyValueIterator, ValueTimestampHeaders> iterator = store.all()) { + mockTime.sleep(3); + } + + assertEquals(2.5 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationAvg.metricValue()); + assertEquals(3.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationMax.metricValue()); + } + + // The above shouldTimeIteratorDuration goes through store.all() -> the KeyValueIterator sibling. + // This pins the same close()-path recording for the ReadOnlyRecordIterator that backs + // TimestampedWindowKeyWithHeadersQuery, whose close() records both the operation sensor (fetch) + // and the iterator-duration sensor via the shared AbstractMeteredIterator lifecycle. + @SuppressWarnings({"unchecked", "rawtypes"}) + @Test + public void shouldTimeIteratorDurationForTimestampedWindowKeyWithHeadersQuery() { + setUp(); + store.init(context, store); + when(innerStoreMock.query(any(), any(PositionBound.class), any(QueryConfig.class))) + .thenReturn( + (QueryResult) QueryResult.forResult(windowKeyIterator(List.of())), + (QueryResult) QueryResult.forResult(windowKeyIterator(List.of()))); + + final KafkaMetric iteratorDurationAvg = metric("iterator-duration-avg"); + final KafkaMetric iteratorDurationMax = metric("iterator-duration-max"); + final KafkaMetric fetchLatencyAvg = metric("fetch-latency-avg"); + assertEquals(Double.NaN, (Double) iteratorDurationAvg.metricValue()); + assertEquals(Double.NaN, (Double) iteratorDurationMax.metricValue()); + + // Two samples (2ms then 3ms) so avg (2.5ms) and max (3ms) differ -- one sample would leave them + // identical and not actually pin avg. Mirrors the sibling shouldTimeIteratorDuration above. + try (ReadOnlyRecordIterator, String> iterator = store.query( + TimestampedWindowKeyWithHeadersQuery.withKeyAndWindowStartRange( + KEY, Instant.ofEpochMilli(5), Instant.ofEpochMilli(100)), + PositionBound.unbounded(), new QueryConfig(false)).getResult()) { + mockTime.sleep(2); + } + + assertEquals(2.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationAvg.metricValue()); + assertEquals(2.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationMax.metricValue()); + + try (ReadOnlyRecordIterator, String> iterator = store.query( + TimestampedWindowKeyWithHeadersQuery.withKeyAndWindowStartRange( + KEY, Instant.ofEpochMilli(5), Instant.ofEpochMilli(100)), + PositionBound.unbounded(), new QueryConfig(false)).getResult()) { + mockTime.sleep(3); + } + + assertEquals(2.5 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationAvg.metricValue()); + assertEquals(3.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationMax.metricValue()); + // fetchSensor is recorded only from the iterator's close() on this path, so the two samples + // (2ms, 3ms) average to exactly 2.5ms. + assertEquals(2.5 * TimeUnit.MILLISECONDS.toNanos(1), (double) fetchLatencyAvg.metricValue()); + } + @SuppressWarnings({"unchecked", "rawtypes"}) @Test public void shouldLeaveIteratorOpenWhenNextThrowsAndNotClosed() { @@ -889,10 +970,14 @@ private static KeyValueIterator, byte[]> windowRangeIterator(fin } private KafkaMetric numOpenIteratorsMetric() { + return metric("num-open-iterators"); + } + + private KafkaMetric metric(final String name) { return metrics.metrics().entrySet().stream() - .filter(entry -> entry.getKey().name().equals("num-open-iterators")) + .filter(entry -> entry.getKey().name().equals(name)) .findFirst() - .orElseThrow(() -> new AssertionError("num-open-iterators metric not registered")) + .orElseThrow(() -> new AssertionError(name + " metric not registered")) .getValue(); } }