nicktelford commented on code in PR #16041:
URL: https://github.com/apache/kafka/pull/16041#discussion_r1613112730
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:
##########
@@ -169,6 +172,10 @@ private void registerMetrics() {
iteratorDurationSensor =
StateStoreMetrics.iteratorDurationSensor(taskId.toString(), metricsScope,
name(), streamsMetrics);
StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(),
metricsScope, name(), streamsMetrics,
(config, now) -> numOpenIterators.get());
+ StateStoreMetrics.addOldestOpenIteratorGauge(taskId.toString(),
metricsScope, name(), streamsMetrics,
+ (config, now) -> openIterators.isEmpty() ? null :
+
openIterators.stream().mapToLong(MeteredIterator::startTimestamp).min().getAsLong()
Review Comment:
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.
:thinking:
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?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
