From 585fc8228d4d590c9b6b8e4adcf289c36c7c2789 Mon Sep 17 00:00:00 2001 From: Jess Jin Date: Tue, 28 Jul 2026 09:47:10 -0400 Subject: [PATCH 1/7] De-duplicate Metered*WithHeaders read-only-record iterators --- ...AbstractMeteredReadOnlyRecordIterator.java | 95 +++++++++++++++++++ .../MeteredSessionStoreWithHeaders.java | 35 +------ ...edTimestampedKeyValueStoreWithHeaders.java | 9 +- ...eredTimestampedWindowStoreWithHeaders.java | 34 +------ 4 files changed, 101 insertions(+), 72 deletions(-) create mode 100644 streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractMeteredReadOnlyRecordIterator.java diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractMeteredReadOnlyRecordIterator.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractMeteredReadOnlyRecordIterator.java new file mode 100644 index 0000000000000..bb0e9256776a0 --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractMeteredReadOnlyRecordIterator.java @@ -0,0 +1,95 @@ +/* + * 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.processor.api.ReadOnlyRecord; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.ReadOnlyRecordIterator; + +import java.util.Set; +import java.util.concurrent.atomic.LongAdder; + +/** + * Shared metering lifecycle for the {@code Metered*WithHeaders} iterators that back the + * headers-aware IQv2 range/window/session query types and yield {@link ReadOnlyRecord}s. + * + *

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. Only {@link #next()} genuinely differs + * per store -- raw key/value types, value deserialization, key derivation, timestamp source, and + * whether a negative/absent timestamp is rejected -- so subclasses implement just that. + * + * @param the raw iterator's key type + * @param the {@link ReadOnlyRecord} key type + * @param the {@link ReadOnlyRecord} value type + */ +abstract class AbstractMeteredReadOnlyRecordIterator + implements ReadOnlyRecordIterator, 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; + + AbstractMeteredReadOnlyRecordIterator(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); + } + + @Override + public long startTimestamp() { + return startTimestampMs; + } + + @Override + public boolean hasNext() { + return iter.hasNext(); + } + + @Override + public 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..e39fc26670acc 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 @@ -585,30 +585,12 @@ 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 AbstractMeteredReadOnlyRecordIterator, Windowed, 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 +608,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..85cdedfb1d568 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 @@ -777,7 +777,7 @@ public K peekNextKey() { * {@code next()} throws. */ private class MeteredTimestampedKeyValueStoreWithHeadersReadOnlyRecordIterator - extends AbstractMeteredIterator implements ReadOnlyRecordIterator { + extends AbstractMeteredReadOnlyRecordIterator { private final Function> valueTimestampHeadersDeserializer; @@ -786,15 +786,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(); 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..69d7062f254cf 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 @@ -662,33 +662,16 @@ public Windowed peekNextKey() { * result, or a {@code Windowed} for a range result */ private class MeteredWindowStoreWithHeadersReadOnlyRecordIterator - implements ReadOnlyRecordIterator, V>, MeteredIterator { + extends AbstractMeteredReadOnlyRecordIterator, 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 +705,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( From 151d3a75c6e6fd1a859d850879c434b28f7d10dd Mon Sep 17 00:00:00 2001 From: Jess Jin Date: Wed, 5 Aug 2026 09:46:09 -0400 Subject: [PATCH 2/7] =?UTF-8?q?MINOR:=20address=20review=20=E2=80=94=20mak?= =?UTF-8?q?e=20metering=20base=20lifecycle-only,=20add=20close-path=20test?= =?UTF-8?q?s?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Broaden the de-duplication per review: rework the shared base into a lifecycle-only AbstractMeteredIterator implements MeteredIterator (no longer binds K/V or a result interface), delete the duplicate inner AbstractMeteredIterator in MeteredTimestampedKeyValueStoreWithHeaders, and have all seven Metered*WithHeaders iterators extend it while declaring their own result interface (ReadOnlyRecordIterator or KeyValueIterator). Add close-path iterator-duration tests: one per test class exercising the ReadOnlyRecordIterator via its query, plus a shouldTimeIteratorDuration for the window store's KeyValueIterator sibling (which had no duration coverage), each asserting the operation and iterator-duration sensors record on close. --- ...ator.java => AbstractMeteredIterator.java} | 40 ++++++----- .../MeteredSessionStoreWithHeaders.java | 33 ++------- ...edTimestampedKeyValueStoreWithHeaders.java | 51 ++------------ ...eredTimestampedWindowStoreWithHeaders.java | 33 ++------- .../MeteredSessionStoreWithHeadersTest.java | 31 ++++++++ ...mestampedKeyValueStoreWithHeadersTest.java | 34 +++++++++ ...TimestampedWindowStoreWithHeadersTest.java | 70 ++++++++++++++++++- 7 files changed, 169 insertions(+), 123 deletions(-) rename streams/src/main/java/org/apache/kafka/streams/state/internals/{AbstractMeteredReadOnlyRecordIterator.java => AbstractMeteredIterator.java} (64%) diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractMeteredReadOnlyRecordIterator.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractMeteredIterator.java similarity index 64% rename from streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractMeteredReadOnlyRecordIterator.java rename to streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractMeteredIterator.java index bb0e9256776a0..d4529ce7d5950 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractMeteredReadOnlyRecordIterator.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractMeteredIterator.java @@ -18,30 +18,30 @@ import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.utils.Time; -import org.apache.kafka.streams.processor.api.ReadOnlyRecord; import org.apache.kafka.streams.state.KeyValueIterator; -import org.apache.kafka.streams.state.ReadOnlyRecordIterator; import java.util.Set; import java.util.concurrent.atomic.LongAdder; /** - * Shared metering lifecycle for the {@code Metered*WithHeaders} iterators that back the - * headers-aware IQv2 range/window/session query types and yield {@link ReadOnlyRecord}s. + * 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. Only {@link #next()} genuinely differs - * per store -- raw key/value types, value deserialization, key derivation, timestamp source, and - * whether a negative/absent timestamp is rejected -- so subclasses implement just that. + * 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 - * @param the {@link ReadOnlyRecord} key type - * @param the {@link ReadOnlyRecord} value type */ -abstract class AbstractMeteredReadOnlyRecordIterator - implements ReadOnlyRecordIterator, MeteredIterator { +abstract class AbstractMeteredIterator implements MeteredIterator { final KeyValueIterator iter; private final Sensor operationSensor; @@ -52,12 +52,12 @@ abstract class AbstractMeteredReadOnlyRecordIterator private final long startNs; private final long startTimestampMs; - AbstractMeteredReadOnlyRecordIterator(final KeyValueIterator iter, - final Sensor operationSensor, - final Sensor iteratorSensor, - final Time time, - final LongAdder numOpenIterators, - final Set openIterators) { + 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; @@ -75,12 +75,14 @@ public long startTimestamp() { return startTimestampMs; } - @Override + /** + * 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(); } - @Override public void close() { try { iter.close(); 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 e39fc26670acc..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,7 +561,8 @@ public Windowed peekNextKey() { * non-negative when the window is constructed, so this iterator's {@code next()} can never throw. */ private class MeteredSessionWithHeadersReadOnlyRecordIterator - extends AbstractMeteredReadOnlyRecordIterator, Windowed, AGG> { + extends AbstractMeteredIterator> + implements ReadOnlyRecordIterator, AGG> { private MeteredSessionWithHeadersReadOnlyRecordIterator( final KeyValueIterator, byte[]> iter 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 85cdedfb1d568..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 AbstractMeteredReadOnlyRecordIterator { + extends AbstractMeteredIterator implements ReadOnlyRecordIterator { private final Function> valueTimestampHeadersDeserializer; @@ -812,7 +773,7 @@ public ReadOnlyRecord next() { } private class MeteredTimestampedKeyValueStoreWithHeadersIterator - extends AbstractMeteredIterator implements KeyValueIterator> { + extends AbstractMeteredIterator implements KeyValueIterator> { private KeyValue> cachedNext; @@ -820,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 69d7062f254cf..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,7 +638,8 @@ public Windowed peekNextKey() { * result, or a {@code Windowed} for a range result */ private class MeteredWindowStoreWithHeadersReadOnlyRecordIterator - extends AbstractMeteredReadOnlyRecordIterator, V> { + extends AbstractMeteredIterator + implements ReadOnlyRecordIterator, V> { private final BiFunction> toWindowedKey; 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..11a5e19ec2ac1 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 @@ -667,6 +667,37 @@ public void shouldTimeIteratorDuration() { assertTrue((Double) iteratorDurationMetric.metricValue() > 0.0); } + // 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(); + + when(innerStore.query(any(), any(PositionBound.class), any(QueryConfig.class))) + .thenReturn((QueryResult) QueryResult.forResult(new KeyValueIteratorStub<>( + Collections., byte[]>>emptyList().iterator()))); + + final KafkaMetric iteratorDurationMetric = metric("iterator-duration-avg"); + final KafkaMetric fetchLatencyMetric = metric("fetch-latency-avg"); + + final QueryResult, String>> result = store.query( + TimestampedWindowRangeWithHeadersQuery.withKey(KEY), + PositionBound.unbounded(), + new QueryConfig(false)); + assertTrue(result.isSuccess()); + try (ReadOnlyRecordIterator, String> iterator = result.getResult()) { + // nothing to iterate; just hold it open, then close + mockTime.sleep(100L); + } + + assertTrue((Double) iteratorDurationMetric.metricValue() > 0.0); + assertTrue((Double) fetchLatencyMetric.metricValue() > 0.0); + } + @Test public void shouldRemoveMetricsOnClose() { setUp(); 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..55c534c65ec30 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,40 @@ 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())); + 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()); + + final QueryResult> result = metered.query( + TimestampedRangeWithHeadersQuery.withNoBounds(), PositionBound.unbounded(), new QueryConfig(false)); + try (ReadOnlyRecordIterator iterator = result.getResult()) { + // nothing to iterate; just hold it open, then close + mockTime.sleep(2); + } + + assertEquals(2.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationAvgMetric.metricValue()); + assertEquals(2.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationMaxMetric.metricValue()); + assertTrue((double) getLatencyAvgMetric.metricValue() > 0.0); + } + @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..0a62249ae5ec3 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,63 @@ 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())); + + 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()); + + try (KeyValueIterator, ValueTimestampHeaders> iterator = store.all()) { + // nothing to iterate; just hold it open, then close + mockTime.sleep(2); + } + + assertEquals(2.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationAvg.metricValue()); + assertEquals(2.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()))); + + 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()); + + final QueryResult, String>> result = store.query( + TimestampedWindowKeyWithHeadersQuery.withKeyAndWindowStartRange( + KEY, Instant.ofEpochMilli(5), Instant.ofEpochMilli(100)), + PositionBound.unbounded(), + new QueryConfig(false)); + try (ReadOnlyRecordIterator, String> iterator = result.getResult()) { + // nothing to iterate; just hold it open, then close + mockTime.sleep(2); + } + + assertEquals(2.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationAvg.metricValue()); + assertEquals(2.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationMax.metricValue()); + assertTrue((double) fetchLatencyAvg.metricValue() > 0.0); + } + @SuppressWarnings({"unchecked", "rawtypes"}) @Test public void shouldLeaveIteratorOpenWhenNextThrowsAndNotClosed() { @@ -889,10 +949,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(); } } From 2edf8f255aef7ef465b740f01aeb6e7b63658001 Mon Sep 17 00:00:00 2001 From: Jess Jin Date: Thu, 6 Aug 2026 09:10:32 -0400 Subject: [PATCH 3/7] =?UTF-8?q?MINOR:=20address=20review=20round=202=20?= =?UTF-8?q?=E2=80=94=20final=20startTimestamp,=20two-sample=20duration=20t?= =?UTF-8?q?ests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Make AbstractMeteredIterator.startTimestamp() final: the constructor's openIterators.add(this) sorts through it via the set's startTimestamp comparator, on a half-built object, so a subclass must not be able to override it with something that reads its own not-yet-assigned state. - Tighten the new iterator-duration tests: open two iterators (2ms then 3ms) and assert exact avg (2.5ms) and max (3ms) instead of one sample / > 0.0, so avg is actually pinned distinctly from max. Applied across the KV, session and window ReadOnlyRecord tests and the window KeyValueIterator sibling test. --- .../internals/AbstractMeteredIterator.java | 5 ++- .../MeteredSessionStoreWithHeadersTest.java | 43 +++++++++++++------ ...mestampedKeyValueStoreWithHeadersTest.java | 20 ++++++--- ...TimestampedWindowStoreWithHeadersTest.java | 39 ++++++++++++----- 4 files changed, 77 insertions(+), 30 deletions(-) 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 index d4529ce7d5950..4f1bd4aab0acd 100644 --- 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 @@ -70,8 +70,11 @@ abstract class AbstractMeteredIterator implements MeteredIterator { 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 long startTimestamp() { + public final long startTimestamp() { return startTimestampMs; } 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 11a5e19ec2ac1..96cd03d367872 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; @@ -678,24 +679,38 @@ public void shouldTimeIteratorDurationForTimestampedWindowRangeWithHeadersQuery( init(); when(innerStore.query(any(), any(PositionBound.class), any(QueryConfig.class))) - .thenReturn((QueryResult) QueryResult.forResult(new KeyValueIteratorStub<>( - Collections., byte[]>>emptyList().iterator()))); - - final KafkaMetric iteratorDurationMetric = metric("iterator-duration-avg"); + .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()); - final QueryResult, String>> result = store.query( - TimestampedWindowRangeWithHeadersQuery.withKey(KEY), - PositionBound.unbounded(), - new QueryConfig(false)); - assertTrue(result.isSuccess()); - try (ReadOnlyRecordIterator, String> iterator = result.getResult()) { - // nothing to iterate; just hold it open, then close - mockTime.sleep(100L); + // 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); } - assertTrue((Double) iteratorDurationMetric.metricValue() > 0.0); - assertTrue((Double) fetchLatencyMetric.metricValue() > 0.0); + 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); + } + + assertEquals(2.5 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationAvgMetric.metricValue()); + assertEquals(3.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationMaxMetric.metricValue()); + assertTrue((double) fetchLatencyMetric.metricValue() > 0.0); } @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 55c534c65ec30..4828bb371b7cb 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 @@ -639,7 +639,9 @@ public void shouldTimeIteratorDuration() { public void shouldTimeIteratorDurationForTimestampedRangeWithHeadersQuery() { setUp(); when(inner.query(any(), any(PositionBound.class), any(QueryConfig.class))) - .thenReturn((QueryResult) QueryResult.forResult(KeyValueIterators.emptyIterator())); + .thenReturn( + (QueryResult) QueryResult.forResult(KeyValueIterators.emptyIterator()), + (QueryResult) QueryResult.forResult(KeyValueIterators.emptyIterator())); init(); final KafkaMetric iteratorDurationAvgMetric = metric("iterator-duration-avg"); @@ -651,15 +653,23 @@ public void shouldTimeIteratorDurationForTimestampedRangeWithHeadersQuery() { assertEquals(Double.NaN, (Double) iteratorDurationAvgMetric.metricValue()); assertEquals(Double.NaN, (Double) iteratorDurationMaxMetric.metricValue()); - final QueryResult> result = metered.query( - TimestampedRangeWithHeadersQuery.withNoBounds(), PositionBound.unbounded(), new QueryConfig(false)); - try (ReadOnlyRecordIterator iterator = result.getResult()) { - // nothing to iterate; just hold it open, then close + // 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()); assertTrue((double) getLatencyAvgMetric.metricValue() > 0.0); } 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 0a62249ae5ec3..a525a11aa15ee 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 @@ -630,20 +630,28 @@ public void shouldDecrementOpenIteratorsTwiceWhenClosedTwiceForTimestampedWindow public void shouldTimeIteratorDuration() { setUp(); store.init(context, store); - when(innerStoreMock.all()).thenReturn(windowRangeIterator(List.of())); + 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()) { - // nothing to iterate; just hold it open, then close 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. @@ -656,7 +664,9 @@ 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()))); + .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"); @@ -664,18 +674,27 @@ public void shouldTimeIteratorDurationForTimestampedWindowKeyWithHeadersQuery() assertEquals(Double.NaN, (Double) iteratorDurationAvg.metricValue()); assertEquals(Double.NaN, (Double) iteratorDurationMax.metricValue()); - final QueryResult, String>> result = store.query( - TimestampedWindowKeyWithHeadersQuery.withKeyAndWindowStartRange( - KEY, Instant.ofEpochMilli(5), Instant.ofEpochMilli(100)), - PositionBound.unbounded(), - new QueryConfig(false)); - try (ReadOnlyRecordIterator, String> iterator = result.getResult()) { - // nothing to iterate; just hold it open, then close + // 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()); assertTrue((double) fetchLatencyAvg.metricValue() > 0.0); } From 228b4dc572a712c47b9f04dc765d2f85f6bc8d08 Mon Sep 17 00:00:00 2001 From: Jess Jin Date: Thu, 6 Aug 2026 09:26:02 -0400 Subject: [PATCH 4/7] MINOR: also two-sample the pre-existing session shouldTimeIteratorDuration Bring the session store's sibling-path duration test in line with the KV one (already two-sample) and the tests added earlier in this PR: open two iterators (2ms then 3ms) and assert exact avg (2.5ms) and max (3ms) instead of a single sample / > 0.0, so avg is pinned distinctly from max. --- .../MeteredSessionStoreWithHeadersTest.java | 34 +++++++++++-------- 1 file changed, 20 insertions(+), 14 deletions(-) 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 96cd03d367872..6705d18755f27 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 @@ -647,25 +647,31 @@ public void shouldTimeIteratorDuration() { setUp(); init(); - final Headers headers = new RecordHeaders(); - headers.add("key1", "value1".getBytes()); - final AggregationWithHeaders valueAndHeaders = AggregationWithHeaders.make(VALUE, headers); - - final AggregationWithHeadersSerializer serializer = new AggregationWithHeadersSerializer<>(Serdes.String().serializer()); - final byte[] serializedValue = serializer.serialize(CHANGELOG_TOPIC, valueAndHeaders); - when(innerStore.fetch(KEY_BYTES)) - .thenReturn(new KeyValueIteratorStub<>( - Collections.singleton(KeyValue.pair(WINDOWED_KEY_BYTES, serializedValue)).iterator())); + .thenReturn( + new KeyValueIteratorStub<>(Collections., byte[]>>emptyList().iterator()), + new KeyValueIteratorStub<>(Collections., byte[]>>emptyList().iterator())); - final KeyValueIterator, AggregationWithHeaders> iterator = store.fetch(KEY); + 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()); - mockTime.sleep(100L); + // 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); + } - iterator.close(); + assertEquals(2.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationAvgMetric.metricValue()); + assertEquals(2.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationMaxMetric.metricValue()); + + try (KeyValueIterator, AggregationWithHeaders> iterator = store.fetch(KEY)) { + 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()); } // The above shouldTimeIteratorDuration goes through store.fetch() -> the KeyValueIterator sibling. From b16f78078407640fe6697f70aba3b184125d3137 Mon Sep 17 00:00:00 2001 From: Jess Jin Date: Thu, 6 Aug 2026 14:41:28 -0400 Subject: [PATCH 5/7] =?UTF-8?q?MINOR:=20address=20review=20round=203=20?= =?UTF-8?q?=E2=80=94=20pin=20operation-sensor=20latency,=20final=20close()?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Assert the operation sensor's latency (get-latency-avg / fetch-latency-avg) is exactly 2.5ms in the three ReadOnlyRecord duration tests instead of > 0.0: that sensor is recorded only from the iterator's close() on these paths, so the two samples (2ms, 3ms) average deterministically. - Make AbstractMeteredIterator.close() final, like startTimestamp(): it owns the metering lifecycle (sensor recording, numOpenIterators decrement, openIterators deregistration), so a subclass overriding it and forgetting super.close() would silently drop the decrement and deregistration. --- .../streams/state/internals/AbstractMeteredIterator.java | 5 ++++- .../state/internals/MeteredSessionStoreWithHeadersTest.java | 4 +++- .../MeteredTimestampedKeyValueStoreWithHeadersTest.java | 4 +++- .../MeteredTimestampedWindowStoreWithHeadersTest.java | 4 +++- 4 files changed, 13 insertions(+), 4 deletions(-) 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 index 4f1bd4aab0acd..af815d730f98b 100644 --- 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 @@ -86,7 +86,10 @@ public boolean hasNext() { return iter.hasNext(); } - public void close() { + // 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 { 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 6705d18755f27..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 @@ -716,7 +716,9 @@ public void shouldTimeIteratorDurationForTimestampedWindowRangeWithHeadersQuery( assertEquals(2.5 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationAvgMetric.metricValue()); assertEquals(3.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationMaxMetric.metricValue()); - assertTrue((double) fetchLatencyMetric.metricValue() > 0.0); + // 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 4828bb371b7cb..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 @@ -670,7 +670,9 @@ public void shouldTimeIteratorDurationForTimestampedRangeWithHeadersQuery() { assertEquals(2.5 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationAvgMetric.metricValue()); assertEquals(3.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationMaxMetric.metricValue()); - assertTrue((double) getLatencyAvgMetric.metricValue() > 0.0); + // 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") 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 a525a11aa15ee..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 @@ -695,7 +695,9 @@ public void shouldTimeIteratorDurationForTimestampedWindowKeyWithHeadersQuery() assertEquals(2.5 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationAvg.metricValue()); assertEquals(3.0 * TimeUnit.MILLISECONDS.toNanos(1), (double) iteratorDurationMax.metricValue()); - assertTrue((double) fetchLatencyAvg.metricValue() > 0.0); + // 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"}) From ad694cf082b020ba2e43a2d05e4028c13aa7fbf3 Mon Sep 17 00:00:00 2001 From: Jess Jin Date: Tue, 11 Aug 2026 11:43:44 -0400 Subject: [PATCH 6/7] trigger ci From 22e922058083679b655193f9e00272fdeff50f15 Mon Sep 17 00:00:00 2001 From: Jess Jin Date: Tue, 11 Aug 2026 11:44:20 -0400 Subject: [PATCH 7/7] trigger ci