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.
--
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
