You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "nicktelford (via GitHub)" <gi...@apache.org> on 2024/11/07 13:26:03 UTC
[PR] KAFKA-17954: Error getting oldest-iterator-open-since-ms from JMX [kafka]
nicktelford opened a new pull request, #17713:
URL: https://github.com/apache/kafka/pull/17713
The thread that evaluates the Gauge for the
oldest-iterator-open-since-ms runs concurrently with threads that
open/close Iterators (stream threads and interactive query threads).
It's therefore possible that between evaluating
`openIterators.isEmpty()` as `false` (because there's at least one open
Iterator) and evaluating `openIterators.first()`, that the last open
Iterator has been closed by a concurrent thread, causing
`openIterators.first()` to throw a `NoSuchElementException`, which will
break the Gauge and potentially the metrics reporting of the
application.
Unfortunately, `ConcurrentSkipListSet` doesn't provide an alternative to
`first()` that retrieves the first element _without removing it_ or
returns `null`. The best we can do is catch the `NoSuchElementException`
and return a `null` instead.
We keep the `openIterators.isEmpty()` check to avoid throwing an
exception in the common-case that there are no open Iterators, because
generating the stack trace when throwing the exception is fairly
expensive.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17954: Error getting oldest-iterator-open-since-ms from JMX [kafka]
Posted by "nicktelford (via GitHub)" <gi...@apache.org>.
nicktelford commented on code in PR #17713:
URL: https://github.com/apache/kafka/pull/17713#discussion_r1836444020
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:
##########
@@ -154,7 +155,13 @@ private void registerMetrics() {
StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
(config, now) -> numOpenIterators.sum());
StateStoreMetrics.addOldestOpenIteratorGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
- (config, now) -> openIterators.isEmpty() ? null : openIterators.first().startTimestamp()
+ (config, now) -> {
+ try {
+ return openIterators.isEmpty() ? null : openIterators.first().startTimestamp();
Review Comment:
> Maybe you can just write your own version of ConcurrentSkipListSet for these iterators, that wraps the ConcurrentSkipListMap while the providing the needed APIs, and hides all the dummy value stuff so it doesn't get too messy in the code itself?
Unfortunately, the internals of `ConcurrentSkipListSet` are `private`, so our own version would need to be a complete copy of the JDK implementation. This is probably not a good idea, as upgrading JDK would not benefit from any improvements in the JDK implementation.
It's not really possible to implement only a subset of the functionality, because we'd still want to implement the same interfaces (namely, `NavigableSet` and `Set`.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17954: Error getting oldest-iterator-open-since-ms from JMX [kafka]
Posted by "nicktelford (via GitHub)" <gi...@apache.org>.
nicktelford commented on code in PR #17713:
URL: https://github.com/apache/kafka/pull/17713#discussion_r1833719869
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:
##########
@@ -154,7 +155,13 @@ private void registerMetrics() {
StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
(config, now) -> numOpenIterators.sum());
StateStoreMetrics.addOldestOpenIteratorGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
- (config, now) -> openIterators.isEmpty() ? null : openIterators.first().startTimestamp()
+ (config, now) -> {
+ try {
+ return openIterators.isEmpty() ? null : openIterators.first().startTimestamp();
Review Comment:
Actually, another option is to use `ConcurrentSkipListMap<MeteredIterator, Object>`. This is actually what `ConcurrentSkipListSet` uses internally, and `ConcurrentSkipListMap` provides `firstEntry()`, which returns `null` when empty.
The downside of this is that everwhere that adds to `openIterators` would need to use a dummy value (`ConcurrentSkipListSet` uses `Boolaen.TRUE`), which might look a bit odd.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17954: Error getting oldest-iterator-open-since-ms from JMX [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17713:
URL: https://github.com/apache/kafka/pull/17713#discussion_r1833587067
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:
##########
@@ -154,7 +155,13 @@ private void registerMetrics() {
StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
(config, now) -> numOpenIterators.sum());
StateStoreMetrics.addOldestOpenIteratorGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
- (config, now) -> openIterators.isEmpty() ? null : openIterators.first().startTimestamp()
+ (config, now) -> {
+ try {
+ return openIterators.isEmpty() ? null : openIterators.first().startTimestamp();
Review Comment:
\cc @ableegoldman @cadonna @lucasbru @bbejeck @aliehsaeedii for feedback? No solution seems to be satisfying.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17954: Error getting oldest-iterator-open-since-ms from JMX [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17713:
URL: https://github.com/apache/kafka/pull/17713#discussion_r1833586372
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:
##########
@@ -154,7 +155,13 @@ private void registerMetrics() {
StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
(config, now) -> numOpenIterators.sum());
StateStoreMetrics.addOldestOpenIteratorGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
- (config, now) -> openIterators.isEmpty() ? null : openIterators.first().startTimestamp()
+ (config, now) -> {
+ try {
+ return openIterators.isEmpty() ? null : openIterators.first().startTimestamp();
Review Comment:
Or we need to use `synchronized(openIterators)` everywhere... -- I don't think we are on the hot code path, so might be ok?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17954: Error getting oldest-iterator-open-since-ms from JMX [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #17713:
URL: https://github.com/apache/kafka/pull/17713#issuecomment-2484771750
Merged to `trunk` and cherry-picked to `3.9` and `3.8` branches.
Thanks for the bug report and fix Nick!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17954: Error getting oldest-iterator-open-since-ms from JMX [kafka]
Posted by "nicktelford (via GitHub)" <gi...@apache.org>.
nicktelford commented on code in PR #17713:
URL: https://github.com/apache/kafka/pull/17713#discussion_r1836444020
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:
##########
@@ -154,7 +155,13 @@ private void registerMetrics() {
StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
(config, now) -> numOpenIterators.sum());
StateStoreMetrics.addOldestOpenIteratorGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
- (config, now) -> openIterators.isEmpty() ? null : openIterators.first().startTimestamp()
+ (config, now) -> {
+ try {
+ return openIterators.isEmpty() ? null : openIterators.first().startTimestamp();
Review Comment:
> Maybe you can just write your own version of ConcurrentSkipListSet for these iterators, that wraps the ConcurrentSkipListMap while the providing the needed APIs, and hides all the dummy value stuff so it doesn't get too messy in the code itself?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17954: Error getting oldest-iterator-open-since-ms from JMX [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #17713:
URL: https://github.com/apache/kafka/pull/17713
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17954: Error getting oldest-iterator-open-since-ms from JMX [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17713:
URL: https://github.com/apache/kafka/pull/17713#discussion_r1833584723
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:
##########
@@ -154,7 +155,13 @@ private void registerMetrics() {
StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
(config, now) -> numOpenIterators.sum());
StateStoreMetrics.addOldestOpenIteratorGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
- (config, now) -> openIterators.isEmpty() ? null : openIterators.first().startTimestamp()
+ (config, now) -> {
+ try {
+ return openIterators.isEmpty() ? null : openIterators.first().startTimestamp();
Review Comment:
Ah. Good point. Missed this...
For this case, does it even make sense to check for `isEmpty()`? We would also just to `return openIterators.first().startTimestamp();` only?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17954: Error getting oldest-iterator-open-since-ms from JMX [kafka]
Posted by "nicktelford (via GitHub)" <gi...@apache.org>.
nicktelford commented on code in PR #17713:
URL: https://github.com/apache/kafka/pull/17713#discussion_r1833342813
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:
##########
@@ -154,7 +155,13 @@ private void registerMetrics() {
StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
(config, now) -> numOpenIterators.sum());
StateStoreMetrics.addOldestOpenIteratorGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
- (config, now) -> openIterators.isEmpty() ? null : openIterators.first().startTimestamp()
+ (config, now) -> {
+ try {
+ return openIterators.isEmpty() ? null : openIterators.first().startTimestamp();
Review Comment:
Unfortunately, that method removes the element from the Set, which we don't want to do in the Gauge!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17954: Error getting oldest-iterator-open-since-ms from JMX [kafka]
Posted by "nicktelford (via GitHub)" <gi...@apache.org>.
nicktelford commented on PR #17713:
URL: https://github.com/apache/kafka/pull/17713#issuecomment-2462773960
@mjsax
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17954: Error getting oldest-iterator-open-since-ms from JMX [kafka]
Posted by "nicktelford (via GitHub)" <gi...@apache.org>.
nicktelford commented on code in PR #17713:
URL: https://github.com/apache/kafka/pull/17713#discussion_r1833711048
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:
##########
@@ -154,7 +155,13 @@ private void registerMetrics() {
StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
(config, now) -> numOpenIterators.sum());
StateStoreMetrics.addOldestOpenIteratorGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
- (config, now) -> openIterators.isEmpty() ? null : openIterators.first().startTimestamp()
+ (config, now) -> {
+ try {
+ return openIterators.isEmpty() ? null : openIterators.first().startTimestamp();
Review Comment:
@mjsax I know it's a bit ugly, but I think this is the best we can do.
`synchronized (openIterators)` is definitely worse, especially in the presence of multiple query threads, but even with just a stream thread, the metrics reporting thread would impact its performance. Throwing an Exception may not be massively fast, but it's only impacting the metrics reporting thread, and we only throw one in the rare case that there's a race between the `isEmpty()` check and `first()`.
> For this case, does it even make sense to check for isEmpty()? We would also just to return openIterators.first().startTimestamp(); only?
We still want to check `isEmpty()` first, simply to avoid having to throw the Exception in the common case that the Set is empty, and we're not racing. The Exception becomes a guard against a race between the `isEmpty()` and `first()` invocations, which are pretty rare, but occasionally happen.
> Or use pollFirst() and re-insert afterwards (also not good code TBH...)
This would cause a race with the stream threads/query threads, because when the Iterator is closed, they remove it from the `openIterators` Set. If this happens while a metrics reporter thread has temporarily removed the Iterator from the Set, it would fail to be removed, and then the reporter thread would add it back, causing the Iterator to leak!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17954: Error getting oldest-iterator-open-since-ms from JMX [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17713:
URL: https://github.com/apache/kafka/pull/17713#discussion_r1833310356
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:
##########
@@ -154,7 +155,13 @@ private void registerMetrics() {
StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
(config, now) -> numOpenIterators.sum());
StateStoreMetrics.addOldestOpenIteratorGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
- (config, now) -> openIterators.isEmpty() ? null : openIterators.first().startTimestamp()
+ (config, now) -> {
+ try {
+ return openIterators.isEmpty() ? null : openIterators.first().startTimestamp();
Review Comment:
Not sure if we need to try-catch... What about using `pollFirst()` ?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17954: Error getting oldest-iterator-open-since-ms from JMX [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #17713:
URL: https://github.com/apache/kafka/pull/17713#discussion_r1835228669
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:
##########
@@ -154,7 +155,13 @@ private void registerMetrics() {
StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
(config, now) -> numOpenIterators.sum());
StateStoreMetrics.addOldestOpenIteratorGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
- (config, now) -> openIterators.isEmpty() ? null : openIterators.first().startTimestamp()
+ (config, now) -> {
+ try {
+ return openIterators.isEmpty() ? null : openIterators.first().startTimestamp();
Review Comment:
Maybe you can just write your own version of `ConcurrentSkipListSet` for these iterators, that wraps the `ConcurrentSkipListMap` while the providing the needed APIs, and hides all the dummy value stuff so it doesn't get too messy in the code itself?
I honestly don't think the current proposal (just catching the exceptions) is that bad, but I understand wanting something better
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17954: Error getting oldest-iterator-open-since-ms from JMX [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #17713:
URL: https://github.com/apache/kafka/pull/17713#discussion_r1849779903
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:
##########
@@ -154,7 +155,13 @@ private void registerMetrics() {
StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
(config, now) -> numOpenIterators.sum());
StateStoreMetrics.addOldestOpenIteratorGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
- (config, now) -> openIterators.isEmpty() ? null : openIterators.first().startTimestamp()
+ (config, now) -> {
+ try {
+ return openIterators.isEmpty() ? null : openIterators.first().startTimestamp();
Review Comment:
> Unfortunately, the internals of ConcurrentSkipListSet are private, so our own version would need to be a complete copy of the JDK implementation. This is probably not a good idea, as upgrading JDK would not benefit from any improvements in the JDK implementation.
I actually think this wouldn't be so bad, since as you pointed out, the ConcurrentSkipListSet is basically a no-op wrapper around the ConcurrentSkipListMap which is where all the messy stuff happens. You could more or less copy the existing ConcurrentSkipListSet implementation as-is and just change the name (and add the `#firstEntry` API we need), it truly does nothing but delegate to the underlying wrap. So I can't imagine there would be anything to change, much less improve, in the JDK implementation of ConcurrentSkipListSet
Just wanted to clarify, as I said before I'm actually fine with the way it's implemented now so I'm not asking for anything to be changed. But if we ever do want to change this or run into a similar problem, imo writing our own wrapper around the ConcurrentSkipListMap is the way to go 🤷♀️
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17954: Error getting oldest-iterator-open-since-ms from JMX [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17713:
URL: https://github.com/apache/kafka/pull/17713#discussion_r1839152414
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:
##########
@@ -154,7 +155,13 @@ private void registerMetrics() {
StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
(config, now) -> numOpenIterators.sum());
StateStoreMetrics.addOldestOpenIteratorGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
- (config, now) -> openIterators.isEmpty() ? null : openIterators.first().startTimestamp()
+ (config, now) -> {
+ try {
+ return openIterators.isEmpty() ? null : openIterators.first().startTimestamp();
Review Comment:
Thanks for the details/background Nick.
If nobody objects, I can go ahead a merge this PR as-is in the next days (just waiting to give people so more time to chime in).
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17954: Error getting oldest-iterator-open-since-ms from JMX [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17713:
URL: https://github.com/apache/kafka/pull/17713#discussion_r1833584723
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:
##########
@@ -154,7 +155,13 @@ private void registerMetrics() {
StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
(config, now) -> numOpenIterators.sum());
StateStoreMetrics.addOldestOpenIteratorGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
- (config, now) -> openIterators.isEmpty() ? null : openIterators.first().startTimestamp()
+ (config, now) -> {
+ try {
+ return openIterators.isEmpty() ? null : openIterators.first().startTimestamp();
Review Comment:
Ah. Good point. Missed this...
For this case, does it even make sense to check for `isEmpty()`? We would also just to `return openIterators.first().startTimestamp();` only?
Or use `pollFirst()` and re-insert afterwards (also not good code TBH...)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17954: Error getting oldest-iterator-open-since-ms from JMX [kafka]
Posted by "nicktelford (via GitHub)" <gi...@apache.org>.
nicktelford commented on code in PR #17713:
URL: https://github.com/apache/kafka/pull/17713#discussion_r1833711048
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:
##########
@@ -154,7 +155,13 @@ private void registerMetrics() {
StateStoreMetrics.addNumOpenIteratorsGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
(config, now) -> numOpenIterators.sum());
StateStoreMetrics.addOldestOpenIteratorGauge(taskId.toString(), metricsScope, name(), streamsMetrics,
- (config, now) -> openIterators.isEmpty() ? null : openIterators.first().startTimestamp()
+ (config, now) -> {
+ try {
+ return openIterators.isEmpty() ? null : openIterators.first().startTimestamp();
Review Comment:
@mjsax I know it's a bit ugly, but I think this is the best we can do.
`synchronized (openIterators)` is definitely worse, especially in the presence of multiple query threads, but even with just a stream thread, the metrics reporting thread would impact its performance. Throwing an Exception may not be massively fast, but it's only impacting the metrics reporting thread, and we only throw one in the rare case that there's a race between the `isEmpty()` check and `first()`.
> For this case, does it even make sense to check for isEmpty()? We would also just to return openIterators.first().startTimestamp(); only?
We still want to check `isEmpty()` first, simply to avoid having to throw the Exception in the common case that the Set is empty. The Exception becomes a guard against a race between the `isEmpty()` and `first()` invocations, which are pretty rare, but occasionally happen.
> Or use pollFirst() and re-insert afterwards (also not good code TBH...)
This would cause a race with the stream threads/query threads, because when the Iterator is closed, they remove it from the `openIterators` Set. If this happens while a metrics reporter thread has temporarily removed the Iterator from the Set, it would fail to be removed, and then the reporter thread would add it back, causing the Iterator to leak!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org