KAFKA-15541: Add oldest-iterator-open-since-ms metric

[nicktelford]

Part of KIP-989

This new StateStore metric tracks the timestamp that the oldest surviving Iterator was created.

This timestamp should continue to climb, and closely track the current time, as old iterators are closed and new ones created. If the timestamp remains very low (i.e. old), that suggests an Iterator has leaked, which should enable users to isolate the affected store.

It will report no data when there are no currently open Iterators.

[nicktelford]

@mjsax This is the final PR to add the KIP-989 metrics. Although I'm considering opening two, much smaller PRs:

1. To add some additional tests to ensure we cover all the Iterator constructors
2. To eliminate some redundancy now all the metrics have landed. Notably, we can replace the AtomicInteger used by num-open-iterators with just counting the cardinality of the Set<MeteredIterator> that we maintain for oldest-iterator-open-since-ms.

[mjsax]

A few questions/suggestions.

Doing follow-up PRs sounds great! We can merge these ofter feature freeze too (we can always add more tests :))

[mjsax]

I don't want to over-engineer (given that we can safely assume that the openIterator set should be small), but wondering if this is the best implementation?

In the end, we only want to track the create ts, not the iterators themselves. And for create ts we could just maintain a list if longs, and we would return list.first() here, and always append to the end of the list when a new iterator is created? Only "remove" would be more expensive, but we could use a sorted tree for the list, and thus remove would be O(log n) not O(n)).

For this case, we also don't need the MeteredIterator helper interface.

Thoughts?

[nicktelford]

Yeah, I considered something like this, but I was concerned that remove could be a problem, in particular for two Iterators with the same timestamp.

An alternative would be to use a ConcurrentSkipListMap<KeyValueIterator<?, ?>, Long>, which would be O(log(n)) for both insert and remove, and eliminate any concerns about duplicate timestamps; whilst still being sorted by timestamp (given the appropriate comparator on construction).

This also solves the problem that this structure should probably be thread-safe, because Iterators may be opened/closed from various Interactive Query threads.

Unfortunately, in Java 8 (the minimum currently supported version), ConcurrentSkipListMap#size() is O(n), and considered "inaccurate". This wasn't addressed until JDK 10, so we will need to keep the AtomicInteger for tracking the num-open-iterators, for the time-being.

Actually, this wouldn't work because the Comparator used by ConcurrentSkipListMap operates on the key, not the value.

🤔

Using ConcurrentSkipListSet instead of HashSet would give us O(1) insert/remove and let us use openIterators.first().startTimestamp() to generate the metric, which is also O(1). It also gives us thread-safety.

Unfortunately, it doesn't let us eliminate the MeteredIterator.

Any better ideas?

[nicktelford]

I've pushed a commit with the switch to ConcurrentSkipListSet, but I can change it again if we come up with something better.

[mjsax]

in particular for two Iterators with the same timestamp.

Yes. We something that allows for duplicates... Not sure if there is any Java standard library impl... Some people implement a custom multi-set on top of tree-map: https://stackoverflow.com/questions/12565587/does-java-have-a-multiset-data-structure-like-the-one-in-c-stl (not sure if it better?) -- seems to be similar to what you propose with ConcurrentSkipListSet?

[mjsax]

This close() call should go into a finally block IMHO:

KeyValueIterator<String, String> second = null;
final long secondTimestamp;
try {
    try (final KeyValueIterator<String, String> first = metered.all()) {
        ...
    }
    
    // now that the first iterator is closed, check that the timestamp has advanced to the still open second iterator
    assertThat((Long) oldestIteratorTimestampMetric.metricValue(), equalTo(secondTimestamp));
} finally {
    if (second != null) {
        second.close();
    }
}

[nicktelford]

Done. I assumed you were suggesting this in all the tests, so I updated them all.

[mjsax]

I think it would be good to get rid of MeteredIterator, but also don't see a good solution w/o over-engineering...

The current ConcurrentSkipListSet is for sure an improvement over the original version. Happy to merge this. Just a nit on the test code.

[nicktelford]

@mjsax Not sure why the build is failing. :streams:compileJava works fine locally. Perhaps the build got confused, can we repeat it?

[mjsax]

[2024-05-28T11:28:17.911Z] > Task :streams:compileJava
[2024-05-28T11:28:17.911Z] /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-16041/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredMultiVersionedKeyQueryIterator.java:54: warning: [this-escape] possible 'this' escape before subclass is fully initialized
[2024-05-28T11:28:17.911Z]         openIterators.add(this);
[2024-05-28T11:28:17.911Z]                           ^
[2024-05-28T11:28:17.911Z] error: warnings found and -Werror specified

I don't expect it to go away... Maybe Java 21 has some stricter/smarter checks?

[mjsax]

Thanks for the KIP @nicktelford -- this will be super useful in the future!