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.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
