KAFKA-19679: Fix NoSuchElementException in oldest open iterator metric #20512

Nikita-Shupletsov · 2025-09-09T16:28:42Z

Querying the oldest-open-iterator metric can result in a
NoSuchElementException when the last open iterator gets removed, due to
a race condition between the query and the metric update.

To avoid this race condition, this PR caches the metric result, to avoid
accessing the list of open iterator directly. We don't need to clear
this cache, because the entire metric is removed when the last iterator
gets removed.

Reviewers: Matthias J. Sax matthias@confluent.io

mjsax · 2025-09-09T16:41:43Z

streams/src/main/java/org/apache/kafka/streams/internals/metrics/OpenIterators.java

-                (config, now) -> openIterators.first().startTimestamp()
+                (config, now) -> {
+                    final Iterator<MeteredIterator> openIteratorsIterator = openIterators.iterator();
+                    return openIteratorsIterator.hasNext() ? openIteratorsIterator.next().startTimestamp() : null;


I think we we ever want to return null -- that why we added dynamic register/deregister behavior.

I believe, we would need to somehow cache the "last result" before we remove the last iterator.

good suggestion. I've updated the PR. please let me know if the changes make sense to you.

…to avoid NoSuchElementException

mjsax

LGTM, with two minor suggestion. Will merge both of them, and merge this PR to get it into AK 4.0.1 release.

streams/src/test/java/org/apache/kafka/streams/internals/metrics/OpenIteratorsTest.java

…cs/OpenIteratorsTest.java

streams/src/test/java/org/apache/kafka/streams/internals/metrics/OpenIteratorsTest.java

…cs/OpenIteratorsTest.java Fix spotless error

#20512) Querying the oldest-open-iterator metric can result in a NoSuchElementException when the last open iterator gets removed, due to a race condition between the query and the metric update. To avoid this race condition, this PR caches the metric result, to avoid accessing the list of open iterator directly. We don't need to clear this cache, because the entire metric is removed when the last iterator gets removed. Reviewers: Matthias J. Sax <matthias@confluent.io>

mjsax · 2025-09-10T04:26:21Z

Thanks for the PR @Nikita-Shupletsov! Merged to trunk and cherry-picked to 4.1 branch.

Congrats to your first contribution!

mjsax · 2025-09-10T04:34:19Z

Oh. While trying to cherry-pick to 4.0 branch, I realize that 4.0.0 is not affected. We never cherry-picked to fix the "return null" issue (https://issues.apache.org/jira/browse/KAFKA-19398) to 4.0.

So if we would want to fix in 4.0.1, we would need to cherry-pick both commits. Might not be worth the trouble, given that a next RC is already lines up.

stevenschlansker

Thank you for taking a look at this issue! I agree this PR fixes the issue as submitted - one comment regarding a possible related issue. We will test this out this week and report any issues we find.

stevenschlansker · 2025-09-10T16:21:53Z

streams/src/main/java/org/apache/kafka/streams/internals/metrics/OpenIterators.java


    public void add(final MeteredIterator iterator) {
        openIterators.add(iterator);
+        updateOldestStartTimestamp();


Possible nano-optimization: the oldest start timestamp will only change if this is the first open iterator - any other open iterator will definitionally be older - so we could possibly only update the oldest timestamp inside the if (openIterators.size() == 1) block below?
(assuming timestamps are all coming from the same time source)

I am not 100% sure I understand the idea.
If we add 3 iterators with timestamps: 1, 2, 3
the result will be 1.
if we delete the first one, the result will be 2. the result changed, even though we still have more than 1 open iterator.

if I misunderstood the idea, please let me know

I am only referring to the add() flow - which will add 1, then add 2, then add 3 - and only the first operation may change the oldest iterator.
I agree in the remove() case we must check every time.

While I was reviewing the PR, I also though about this, but though it's not worth doing, as creating iterators should not happen with very high frequency?

In practice, I doubt it matters much, which is why I called it a "nano" optimization. I leave it in your capable hands to decide if it is worth implementing :)

That said, we actually are having scalability issues regarding creating iterators at relatively high frequency - see https://issues.apache.org/jira/browse/KAFKA-19678

One of our data stores is used almost exclusively for range lookups which use iterators.

stevenschlansker · 2025-09-10T16:26:25Z

streams/src/main/java/org/apache/kafka/streams/internals/metrics/OpenIterators.java

        openIterators.add(iterator);
+        updateOldestStartTimestamp();

        if (openIterators.size() == 1) {


Thinking about this more, I wonder if there is another possible race here that would not cause an exception but could cause missing or incorrectly persisting metrics. The OpenIterators tracker itself is not synchronized (good to avoid contention) but that means that we are mutating add() or remove() and calling size() expecting the results in program order.
But, I don't see any guarantee this is the case - my (limited) understanding is that OpenIterators is a property of a KeyValueStore, which may have partitions shared between threads.

Imagine:

T1: open iterator on store S1 (partition 0) T2: open iterator on store S1 (partition 1) T1: openIterators.add(it1) T2: openIterators.add(it2) T1: if (openIterators.size() == 1) { (2 == 1? nope) T2: if (openIterators.size() == 1) { (2 == 1? nope) End: 2 iterators open, no metric registered

Same race is possible with remove.

I agree. there is a potential race condition here. I will take a closer look

I don't think that's a problem. While stores are sharded, each shard is its own StateStore and thus has an own OpenIterator instance. Hence, OpenIterator should be used by a single StreamsThreads and there should not be any concurrent usage.

The concurrently issue we fixed in only for the metric, because addOldestOpenIteratorGauge registers the Gauge (config, now) -> openIterators.first().startTimestamp() (well, now (config, now) -> oldestStartTimestamp.get()) which could be used concurrently, when the metric is read.

OK, thanks for clarifying!

github-actions bot added triage PRs from the community streams small Small PRs and removed triage PRs from the community labels Sep 9, 2025

mjsax reviewed Sep 9, 2025

View reviewed changes

KAFKA-19679: Changed the calculation logic of oldest start timestamp …

045efd4

…to avoid NoSuchElementException

Nikita-Shupletsov force-pushed the KAFKA-19679 branch from a2da6c3 to 045efd4 Compare September 9, 2025 19:39

Nikita-Shupletsov changed the title ~~KAFKA-19679: Fixed the regression in the metric calculation to avoid NoSuchElementException~~ KAFKA-19679: Changed the calculation logic of oldest start timestamp to avoid NoSuchElementException Sep 9, 2025

mjsax approved these changes Sep 10, 2025

View reviewed changes

streams/src/test/java/org/apache/kafka/streams/internals/metrics/OpenIteratorsTest.java Outdated Show resolved Hide resolved

streams/src/test/java/org/apache/kafka/streams/internals/metrics/OpenIteratorsTest.java Show resolved Hide resolved

mjsax added 2 commits September 9, 2025 18:18

Update streams/src/test/java/org/apache/kafka/streams/internals/metri…

4b52127

…cs/OpenIteratorsTest.java

Update streams/src/test/java/org/apache/kafka/streams/internals/metri…

42ef9d7

…cs/OpenIteratorsTest.java

mjsax changed the title ~~KAFKA-19679: Changed the calculation logic of oldest start timestamp to avoid NoSuchElementException~~ KAFKA-19679: Fix NoSuchElementException in oldest open iterator metric Sep 10, 2025

mjsax added the ci-approved label Sep 10, 2025

mjsax reviewed Sep 10, 2025

View reviewed changes

streams/src/test/java/org/apache/kafka/streams/internals/metrics/OpenIteratorsTest.java Show resolved Hide resolved

Update streams/src/test/java/org/apache/kafka/streams/internals/metri…

ece6179

…cs/OpenIteratorsTest.java Fix spotless error

mjsax merged commit 8015c87 into apache:trunk Sep 10, 2025
20 checks passed

stevenschlansker reviewed Sep 10, 2025

View reviewed changes

KAFKA-19679: Fix NoSuchElementException in oldest open iterator metric #20512

KAFKA-19679: Fix NoSuchElementException in oldest open iterator metric #20512

Uh oh!

Conversation

Nikita-Shupletsov commented Sep 9, 2025 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

mjsax commented Sep 10, 2025

Uh oh!

mjsax commented Sep 10, 2025

Uh oh!

stevenschlansker left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stevenschlansker Sep 10, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stevenschlansker Sep 10, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Nikita-Shupletsov commented Sep 9, 2025 •

edited by github-actions bot

Loading

stevenschlansker Sep 10, 2025 •

edited

Loading

stevenschlansker Sep 10, 2025 •

edited

Loading