diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStore.java index 72caa56d88e35..e44f2af701fc6 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStore.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStore.java @@ -318,6 +318,7 @@ private MeteredTimestampedKeyValueStoreIterator(final KeyValueIteratorwithNoBounds() + ); + } + + @Test + public void shouldTrackOpenIteratorsMetricForTimestampedRangeQuery() { + assertTracksOpenIteratorsMetricForQuery( + TimestampedRangeQuery.withNoBounds() + ); + } + + @SuppressWarnings({"rawtypes", "unchecked"}) + private void assertTracksOpenIteratorsMetricForQuery(final Query query) { + setUp(); + + final QueryResult> rawResult = + QueryResult.forResult(KeyValueIterators.emptyIterator()); + + when(inner.query( + any(), + any(PositionBound.class), + any(QueryConfig.class) + )).thenReturn((QueryResult) rawResult); + + init(); + + final KafkaMetric openIteratorsMetric = metric("num-open-iterators"); + assertThat(openIteratorsMetric, not(nullValue())); + assertThat((Long) openIteratorsMetric.metricValue(), equalTo(0L)); + + final QueryResult result = metered.query( + query, + PositionBound.unbounded(), + new QueryConfig(false) + ); + + assertTrue(result.isSuccess()); + + try (final KeyValueIterator unused = + (KeyValueIterator) result.getResult()) { + assertThat((Long) openIteratorsMetric.metricValue(), equalTo(1L)); + } + + assertThat((Long) openIteratorsMetric.metricValue(), equalTo(0L)); + } + + @SuppressWarnings("unused") @Test public void shouldTimeIteratorDuration() {