You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "dino2895 (via GitHub)" <gi...@apache.org> on 2026/07/10 22:16:18 UTC
[PR] KAFKA-20796: Fix num-open-iterators metric for timestamped range queries [kafka]
dino2895 opened a new pull request, #22809:
URL: https://github.com/apache/kafka/pull/22809
`MeteredTimestampedKeyValueStoreIterator` registered itself in
`openIterators`, but did not update `numOpenIterators`. As a result,
iterators created by IQv2 `RangeQuery` and `TimestampedRangeQuery`
were not included in the `num-open-iterators` metric.
This change increments the counter when the iterator is created and
decrements it when the iterator is closed.
Regression tests verify that the metric transitions from 0 to 1 while
the iterator is open and returns to 0 after it is closed for both query
types.
Testing:
- `./gradlew :streams:spotlessJavaCheck`
- `./gradlew :streams:checkstyleMain :streams:checkstyleTest`
- `./gradlew :streams:test`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20796: Fix num-open-iterators metric for timestamped range queries [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #22809:
URL: https://github.com/apache/kafka/pull/22809#discussion_r4222075394
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:
##########
@@ -104,9 +98,7 @@ public class MeteredKeyValueStore<K, V>
private TaskId taskId;
private Sensor restoreSensor;
- protected LongAdder numOpenIterators = new LongAdder();
- protected NavigableSet<MeteredIterator> openIterators = new ConcurrentSkipListSet<>(Comparator.comparingLong(MeteredIterator::startTimestamp));
-
+ protected final MeteredIteratorTracker iteratorTracker = new MeteredIteratorTracker();
Review Comment:
package-private is more suitable for this variable
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredIteratorTracker.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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 java.util.Comparator;
+import java.util.Iterator;
+import java.util.NavigableSet;
+import java.util.NoSuchElementException;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.atomic.LongAdder;
+
+final class MeteredIteratorTracker {
+
+ private final LongAdder numOpenIterators;
+ private final NavigableSet<MeteredIterator> openIterators;
+
+ MeteredIteratorTracker() {
+ this(
+ new LongAdder(),
+ new ConcurrentSkipListSet<>(Comparator.comparingLong(MeteredIterator::startTimestamp))
+ );
+ }
+
+ MeteredIteratorTracker(final LongAdder numOpenIterators,
+ final NavigableSet<MeteredIterator> openIterators) {
+ this.numOpenIterators = numOpenIterators;
+ this.openIterators = openIterators;
+ }
+
+ void add(final MeteredIterator iterator) {
+ numOpenIterators.increment();
+ openIterators.add(iterator);
+ }
+
+ void remove(final MeteredIterator iterator) {
+ numOpenIterators.decrement();
+ openIterators.remove(iterator);
+ }
+
+ long numOpenIterators() {
+ return numOpenIterators.sum();
+ }
+
+ long oldestIteratorStartTimestamp() {
+ try {
+ final Iterator<MeteredIterator> iterator = openIterators.iterator();
+ return iterator.hasNext() ? iterator.next().startTimestamp() : 0L;
+ } catch (final NoSuchElementException e) {
Review Comment:
I’m confused why we need this catch. If the set is thread-safe, the iterator prefetches the next item when hasNext is called, IIRC, so next can’t throw `NoSuchElementException`. If it is not thread-safe, it throws `ConcurrentModificationException` instead, which this catch doesn’t cover anyway.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20796: Fix num-open-iterators metric for timestamped range queries [kafka]
Posted by "dino2895 (via GitHub)" <gi...@apache.org>.
dino2895 commented on code in PR #22809:
URL: https://github.com/apache/kafka/pull/22809#discussion_r3715932840
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreTest.java:
##########
@@ -451,6 +457,56 @@ public void shouldTrackOpenIteratorsMetric() {
assertThat((Long) openIteratorsMetric.metricValue(), equalTo(0L));
}
+ @Test
+ public void shouldTrackOpenIteratorsMetricForRangeQuery() {
+ assertTracksOpenIteratorsMetricForQuery(
+ RangeQuery.<String, String>withNoBounds()
+ );
+ }
+
+ @Test
+ public void shouldTrackOpenIteratorsMetricForTimestampedRangeQuery() {
+ assertTracksOpenIteratorsMetricForQuery(
+ TimestampedRangeQuery.<String, String>withNoBounds()
+ );
+ }
+
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private void assertTracksOpenIteratorsMetricForQuery(final Query<?> query) {
+ setUp();
+
+ final QueryResult<KeyValueIterator<Bytes, byte[]>> 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));
+ }
+
+
Review Comment:
Fixed. Thanks for catching this!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20796: Fix num-open-iterators metric for timestamped range queries [kafka]
Posted by "dino2895 (via GitHub)" <gi...@apache.org>.
dino2895 commented on code in PR #22809:
URL: https://github.com/apache/kafka/pull/22809#discussion_r3715936691
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreTest.java:
##########
Review Comment:
Added the corresponding plain-store edge-case tests for closing an iterator twice and for next() throwing without implicitly closing it. Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20796: Fix num-open-iterators metric for timestamped range queries [kafka]
Posted by "Jess668 (via GitHub)" <gi...@apache.org>.
Jess668 commented on code in PR #22809:
URL: https://github.com/apache/kafka/pull/22809#discussion_r3645837290
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeadersTest.java:
##########
Review Comment:
This PR branched before KAFKA-20704([#22770](https://github.com/apache/kafka/pull/22770)) merged. That change already fixed the with-headers store and added tests to `MeteredTimestampedKeyValueStoreWithHeadersTest.java` with the same method names you're adding here (`shouldTrackOpenIteratorsMetricForRangeQuery`, `shouldTrackOpenIteratorsMetricForTimestampedRangeQuery`). Could you rebase on latest trunk?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredIteratorTracker.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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 java.util.Comparator;
+import java.util.Iterator;
+import java.util.NavigableSet;
+import java.util.NoSuchElementException;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.atomic.LongAdder;
+
+final class MeteredIteratorTracker {
+
+ private final LongAdder numOpenIterators = new LongAdder();
+
+ private final NavigableSet<MeteredIterator> openIterators =
+ new ConcurrentSkipListSet<>(
+ Comparator.comparingLong(MeteredIterator::startTimestamp)
+ );
+
+ void add(final MeteredIterator iterator) {
+ numOpenIterators.increment();
+ openIterators.add(iterator);
+ }
+
+ void remove(final MeteredIterator iterator) {
+ numOpenIterators.decrement();
+ openIterators.remove(iterator);
+ }
+
+ long numOpenIterators() {
+ return numOpenIterators.sum();
+ }
+
+ long oldestIteratorStartTimestamp() {
+ try {
+ final Iterator<MeteredIterator> iterator = openIterators.iterator();
+ return iterator.hasNext() ? iterator.next().startTimestamp() : 0L;
+ } catch (final NoSuchElementException e) {
+ return 0L;
+ }
+ }
Review Comment:
Nit: file is missing a trailing newline.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreTest.java:
##########
Review Comment:
Suggestion: mirror the with-headers edge-case coverage for the plain store. Consider adding `shouldDecrementOpenIteratorsTwiceWhenClosedTwiceFor...` and `shouldLeaveIteratorOpenWhenNextThrowsAndNotClosedFor...`, pairing its sibling `MeteredTimestampedKeyValueStoreWithHeadersTest.java`
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreTest.java:
##########
@@ -451,6 +457,56 @@ public void shouldTrackOpenIteratorsMetric() {
assertThat((Long) openIteratorsMetric.metricValue(), equalTo(0L));
}
+ @Test
+ public void shouldTrackOpenIteratorsMetricForRangeQuery() {
+ assertTracksOpenIteratorsMetricForQuery(
+ RangeQuery.<String, String>withNoBounds()
+ );
+ }
+
+ @Test
+ public void shouldTrackOpenIteratorsMetricForTimestampedRangeQuery() {
+ assertTracksOpenIteratorsMetricForQuery(
+ TimestampedRangeQuery.<String, String>withNoBounds()
+ );
+ }
+
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private void assertTracksOpenIteratorsMetricForQuery(final Query<?> query) {
+ setUp();
+
+ final QueryResult<KeyValueIterator<Bytes, byte[]>> 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));
+ }
+
+
Review Comment:
Nit: drop the extra blank line after the new test method
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20796: Fix num-open-iterators metric for timestamped range queries [kafka]
Posted by "dino2895 (via GitHub)" <gi...@apache.org>.
dino2895 commented on PR #22809:
URL: https://github.com/apache/kafka/pull/22809#issuecomment-4963052335
Thanks for the review. Good catch.
```MeteredTimestampedKeyValueStoreWithHeadersQueryIterator``` does have the same issue. ```MeteredMultiVersionedKeyQueryIterator``` already updates both variables correctly, but it would benefit from using the same helper. I’ll introduce a tracker that encapsulates iterator registration and removal, migrate these iterators to it, and add regression tests for the with-headers query paths.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20796: Fix num-open-iterators metric for timestamped range queries [kafka]
Posted by "dino2895 (via GitHub)" <gi...@apache.org>.
dino2895 commented on code in PR #22809:
URL: https://github.com/apache/kafka/pull/22809#discussion_r3715925266
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredIteratorTracker.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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 java.util.Comparator;
+import java.util.Iterator;
+import java.util.NavigableSet;
+import java.util.NoSuchElementException;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.atomic.LongAdder;
+
+final class MeteredIteratorTracker {
+
+ private final LongAdder numOpenIterators = new LongAdder();
+
+ private final NavigableSet<MeteredIterator> openIterators =
+ new ConcurrentSkipListSet<>(
+ Comparator.comparingLong(MeteredIterator::startTimestamp)
+ );
+
+ void add(final MeteredIterator iterator) {
+ numOpenIterators.increment();
+ openIterators.add(iterator);
+ }
+
+ void remove(final MeteredIterator iterator) {
+ numOpenIterators.decrement();
+ openIterators.remove(iterator);
+ }
+
+ long numOpenIterators() {
+ return numOpenIterators.sum();
+ }
+
+ long oldestIteratorStartTimestamp() {
+ try {
+ final Iterator<MeteredIterator> iterator = openIterators.iterator();
+ return iterator.hasNext() ? iterator.next().startTimestamp() : 0L;
+ } catch (final NoSuchElementException e) {
+ return 0L;
+ }
+ }
Review Comment:
Fixed — added the missing trailing newline. Thanks for catching this!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20796: Fix num-open-iterators metric for timestamped range queries [kafka]
Posted by "dino2895 (via GitHub)" <gi...@apache.org>.
dino2895 commented on PR #22809:
URL: https://github.com/apache/kafka/pull/22809#issuecomment-4963595186
Thanks for the suggestion. I introduced `MeteredIteratorTracker` to keep
the open-iterator counter and oldest-iterator tracking set encapsulated
behind a single add/remove call.
I migrated the relevant key-value store iterators, including
`MeteredMultiVersionedKeyQueryIterator`, and fixed the same missing
counter update in
`MeteredTimestampedKeyValueStoreWithHeadersQueryIterator`.
I also added regression tests for the with-headers `RangeQuery` and
`TimestampedRangeQuery` paths, verifying the open-iterator metric
transitions from 0 to 1 and back to 0.
Verification:
- `./gradlew :streams:spotlessJavaCheck`
- `./gradlew :streams:checkstyleMain :streams:checkstyleTest`
- `./gradlew :streams:test`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20796: Fix num-open-iterators metric for timestamped range queries [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on PR #22809:
URL: https://github.com/apache/kafka/pull/22809#issuecomment-5177829208
@dino2895 a gentle reminder!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20796: Fix num-open-iterators metric for timestamped range queries [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #22809:
URL: https://github.com/apache/kafka/pull/22809#issuecomment-5982971356
@dino2895 please rebase. We need another CI run."
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20796: Fix num-open-iterators metric for timestamped range queries [kafka]
Posted by "dino2895 (via GitHub)" <gi...@apache.org>.
dino2895 commented on code in PR #22809:
URL: https://github.com/apache/kafka/pull/22809#discussion_r3715928631
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeadersTest.java:
##########
Review Comment:
Rebased onto the latest trunk and dropped the duplicate with-headers tests, since #22770 already provides the coverage. Thanks for pointing this out!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20796: Fix num-open-iterators metric for timestamped range queries [kafka]
Posted by "dino2895 (via GitHub)" <gi...@apache.org>.
dino2895 commented on code in PR #22809:
URL: https://github.com/apache/kafka/pull/22809#discussion_r4241726887
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:
##########
@@ -104,9 +98,7 @@ public class MeteredKeyValueStore<K, V>
private TaskId taskId;
private Sensor restoreSensor;
- protected LongAdder numOpenIterators = new LongAdder();
- protected NavigableSet<MeteredIterator> openIterators = new ConcurrentSkipListSet<>(Comparator.comparingLong(MeteredIterator::startTimestamp));
-
+ protected final MeteredIteratorTracker iteratorTracker = new MeteredIteratorTracker();
Review Comment:
Fixed. Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20796: Fix num-open-iterators metric for timestamped range queries [kafka]
Posted by "dino2895 (via GitHub)" <gi...@apache.org>.
dino2895 commented on code in PR #22809:
URL: https://github.com/apache/kafka/pull/22809#discussion_r4241725772
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredIteratorTracker.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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 java.util.Comparator;
+import java.util.Iterator;
+import java.util.NavigableSet;
+import java.util.NoSuchElementException;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.atomic.LongAdder;
+
+final class MeteredIteratorTracker {
+
+ private final LongAdder numOpenIterators;
+ private final NavigableSet<MeteredIterator> openIterators;
+
+ MeteredIteratorTracker() {
+ this(
+ new LongAdder(),
+ new ConcurrentSkipListSet<>(Comparator.comparingLong(MeteredIterator::startTimestamp))
+ );
+ }
+
+ MeteredIteratorTracker(final LongAdder numOpenIterators,
+ final NavigableSet<MeteredIterator> openIterators) {
+ this.numOpenIterators = numOpenIterators;
+ this.openIterators = openIterators;
+ }
+
+ void add(final MeteredIterator iterator) {
+ numOpenIterators.increment();
+ openIterators.add(iterator);
+ }
+
+ void remove(final MeteredIterator iterator) {
+ numOpenIterators.decrement();
+ openIterators.remove(iterator);
+ }
+
+ long numOpenIterators() {
+ return numOpenIterators.sum();
+ }
+
+ long oldestIteratorStartTimestamp() {
+ try {
+ final Iterator<MeteredIterator> iterator = openIterators.iterator();
+ return iterator.hasNext() ? iterator.next().startTimestamp() : 0L;
+ } catch (final NoSuchElementException e) {
Review Comment:
This catch was carried over from the existing gauge implementation. The ConcurrentSkipListSet iterator caches the next element, so concurrent removal won't cause next() to throw after hasNext() returns true. I've removed the redundant catch and the unused import.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org