You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "cadonna (via GitHub)" <gi...@apache.org> on 2023/11/13 10:48:01 UTC
[PR] KAFKA-15555: Ensure wakeups are handled correctly in poll() [kafka]
cadonna opened a new pull request, #14746:
URL: https://github.com/apache/kafka/pull/14746
We need to be careful when aborting a long poll with wakeup() since the consumer might never return records if the poll is interrupted after the consumer position has been updated but the records have not been returned to the caller of poll().
This PR avoids wake-ups during this critical period in the PrototypeAsyncConsumer, similarly as in the current consumer.
*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #14746:
URL: https://github.com/apache/kafka/pull/14746#discussion_r1394385232
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -415,6 +439,27 @@ public ConsumerRecords<K, V> poll(final Duration timeout) {
return ConsumerRecords.empty();
} finally {
kafkaConsumerMetrics.recordPollEnd(timer.currentTimeMs());
+ wakeupTrigger.clearActiveTask();
+ }
+ }
+
+ private CompletableFuture<Void> setupWakeupTrigger() {
+ final CompletableFuture<Void> wakeupFuture = new CompletableFuture<>();
+ return wakeupTrigger.setActiveTask(wakeupFuture);
Review Comment:
I had to scratch my head a little bit over this.
Since we already have the `Wakeupable` interface, I wonder if it wouldn't be cleaner to have a new static subclass:
```
/** Placeholder for a task that delays wake-up until it's manually triggered */
DelayedWakeupTask(final boolean wakeUpMarker)
```
Then replace `DelayedWakeupTask(false)` by `DelayedWakeupTask(true)` in `WakeupTrigger.wakeup` and move `maybeTriggerWakeup` to `WakeupTrigger`, rename it to `maybeTriggerDelayedWakeup`.
Then we'd have fewer code paths to worry about (no non-exceptional completion, no other possible exceptions where we ourselves don't know what to do and just throw and `IllegalStateException`).
Its just an idea, could be that I overlooked something.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -168,6 +178,44 @@ public void testCommitted_ExceptionThrown() {
}
}
+ @Test
+ public void testWakeupBeforeCallingPoll() {
+ final TopicPartition tp = new TopicPartition("foo", 3);
+ consumer.assign(singleton(tp));
+
+ consumer.wakeup();
+
+ assertThrows(WakeupException.class, () -> consumer.poll(Duration.ofMinutes(1)));
+ }
+
+ @Test
+ public void testWakeupAfterEmptyFetch() {
Review Comment:
Do we not want to test `wake-up` with a non-empty fetch?
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on PR #14746:
URL: https://github.com/apache/kafka/pull/14746#issuecomment-1816005251
> I wonder if we could just use a BlockingQueue for the fetchBuffer because fetchBuffer.poll(time) blocks until non-empty or timeout.
As far as I see from the javadocs of a `BlockingQueue` there is no way to wake up from `poll()` other than interrupting the thread. I think signalling on a condition variable is safer than interrupting a thread since its scope is well-defined.
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #14746:
URL: https://github.com/apache/kafka/pull/14746#discussion_r1394418301
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -415,6 +439,27 @@ public ConsumerRecords<K, V> poll(final Duration timeout) {
return ConsumerRecords.empty();
} finally {
kafkaConsumerMetrics.recordPollEnd(timer.currentTimeMs());
+ wakeupTrigger.clearActiveTask();
Review Comment:
Hmm, I'm not 100% sure we are providing the guarantee stated in the javadoc of the consumer here.
This would be:
```
* If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
```
But if the current fetch returns records, we will never throw a `WakeupException`, not from this `poll` nor the next. Should we keep the "activeTask" around strictly until we are ready to throw a `WakeupException`?
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14746:
URL: https://github.com/apache/kafka/pull/14746#issuecomment-1809158196
Another point I want to make here is that the wakeup call also wakes-up the blocking client. I wonder if we also need to do that to the network thread - @kirktrue
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on PR #14746:
URL: https://github.com/apache/kafka/pull/14746#issuecomment-1814256948
@philipnee @kirktrue @lucasbru I am concerned that when one calls `wakeup()` the application thread might stay blocked in waiting for a non-empty fetch buffer.
See https://github.com/apache/kafka/blob/a16cde3b1f67ea3248466444fd3562878428fecf/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java#L968
Currently, I do not know how to interrupt the application thread with the current code. I cannot do it in `AsyncKafkaConsumer#wakeup()` because I do not have a handle on the application thread and I cannot do it in `WakeupTrigger#maybeTriggerWakeup()` because there it is too late. Any 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on PR #14746:
URL: https://github.com/apache/kafka/pull/14746#issuecomment-1814405730
Thanks for the options, @lucasbru!
In the meanwhile, I also had an idea to store the reference to the application thread in a wakeupable task similar to your `FetchAction` task. A call to `wakeup()` would then interrupt the application thread stored in the wakeupable task. However, I will try your signal idea since that seems more safe to me then interrupting the application thread.
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #14746:
URL: https://github.com/apache/kafka/pull/14746#issuecomment-1814363083
> @philipnee @kirktrue @lucasbru I am concerned that when one calls `wakeup()` the application thread might stay blocked in waiting for a non-empty fetch buffer.
Yes. The application thread remains blocked, so this doesn't really wake the application thread up.
> Currently, I do not know how to interrupt the application thread with the current code. I cannot do it in `AsyncKafkaConsumer#wakeup()` because I do not have a handle on the application thread and I cannot do it in `WakeupTrigger#maybeTriggerWakeup()` because there it is too late. Any ideas?
I'd say we signal `notEmptyCondition` in `FetchBuffer`. That could possibly be achieved this way:
- We implement a new subclass of `Wakeupable` that is a `FetchAction(FetchBuffer)`, similar to my idea above.
- In `AsynConsumer.poll`, `FetchAction` is registered before the poll and removed after the poll.
- In `WakeupTrigger.wakeup()`, `FetchAction` is replaced by a `WakeupFuture`, and `FetchBuffer.wakeup` is called.
- `FetchBuffer.wakeup` calls `notEmptyCondition.signalAll()`.
There are other ideas, but I think they are more problematic:
1. If we wanted to stick more to the original architecture, we could attempt to solve "waiting for a non-empty fetch buffer" by putting a future in the background queue and let the background thread poll loop complete the future. But that would be a significant rewrite and may have other problems.
2. I'm implementing a change where only one thread can use the consumer API, just like in the original consumer. In principle, you could store a reference to the application thread in the consumer when it is opened and interrupt it that way.
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on PR #14746:
URL: https://github.com/apache/kafka/pull/14746#issuecomment-1812010386
@philipnee @kirktrue Thanks for your comments and explanation! I totally missed that if the future is completed exceptionally it throws an `ExecutionException`. The javadocs of `completeExceptionally()` state:
```
If not already completed, causes invocations of get() and related methods to throw the given exception.
```
It does not say anything about wrapping. I just saw now that the javadocs of `CompletableFuture` state it more clearly:
```
In case of exceptional completion with a CompletionException, methods get() and get(long, TimeUnit) throw an ExecutionException with the same cause as held in the corresponding CompletionException.
```
I misunderstood the javadocs on `get()`. I modified the code to use `getNow()` which makes the handling a bit simpler.
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on PR #14746:
URL: https://github.com/apache/kafka/pull/14746#issuecomment-1815469269
Can we add the `ctr` label here for easier discovery?
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on PR #14746:
URL: https://github.com/apache/kafka/pull/14746#issuecomment-1824090907
Build failures are unrelated.
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #14746:
URL: https://github.com/apache/kafka/pull/14746#discussion_r1397154783
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchBuffer.java:
##########
@@ -52,6 +53,8 @@ public class FetchBuffer implements AutoCloseable {
private final Condition notEmptyCondition;
private final IdempotentCloser idempotentCloser = new IdempotentCloser();
+ private final AtomicBoolean wakedUp = new AtomicBoolean(false);
Review Comment:
Oh, boy! How embarrassing!
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #14746:
URL: https://github.com/apache/kafka/pull/14746#discussion_r1397156239
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchBuffer.java:
##########
@@ -185,6 +188,16 @@ void awaitNotEmpty(Timer timer) {
}
}
+ void wakeup() {
Review Comment:
I used the same format as on the `Consumer` interface. Same for my recent change to `wokenup`.
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #14746:
URL: https://github.com/apache/kafka/pull/14746#discussion_r1394568452
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -168,6 +178,44 @@ public void testCommitted_ExceptionThrown() {
}
}
+ @Test
+ public void testWakeupBeforeCallingPoll() {
+ final TopicPartition tp = new TopicPartition("foo", 3);
+ consumer.assign(singleton(tp));
+
+ consumer.wakeup();
+
+ assertThrows(WakeupException.class, () -> consumer.poll(Duration.ofMinutes(1)));
+ }
+
+ @Test
+ public void testWakeupAfterEmptyFetch() {
Review Comment:
I forgot about that unit test! Thanks!
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna merged PR #14746:
URL: https://github.com/apache/kafka/pull/14746
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on PR #14746:
URL: https://github.com/apache/kafka/pull/14746#issuecomment-1807924858
I will add tests after somebody confirms that this makes sense.
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14746:
URL: https://github.com/apache/kafka/pull/14746#issuecomment-1810853645
Basically future.get() API only return 3 types of exceptions: ExecutionException, InterruptedException, and Cancellation per documentation.
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14746:
URL: https://github.com/apache/kafka/pull/14746#issuecomment-1809152957
Hi @cadonna - Thank you for putting time into this PR. Based on my understanding this PR does 2 things: if wakeup() is invoked before calling poll(), the consumer will return immediately. If wakeup() is invoked during poll(), we should get a wakeupException and return. Overall I think it looks right.
*while writing this I think @kirktrue has asked the questions I wanted to ask.
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #14746:
URL: https://github.com/apache/kafka/pull/14746#discussion_r1390948525
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java:
##########
@@ -335,10 +336,17 @@ public ConsumerRecords<K, V> poll(final Duration timeout) {
throw new IllegalStateException("Consumer is not subscribed to any topics or assigned any partitions");
}
+ final CompletableFuture<Void> wakeupFuture = setupWakeupTrigger();
Review Comment:
I needed to create a dedicated future since fetches are not passed to the background thread as events since fetches happen continuously on the background thread.
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14746:
URL: https://github.com/apache/kafka/pull/14746#issuecomment-1815756879
@cadonna - There are some build failures. Can we rerun the tests?
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14746:
URL: https://github.com/apache/kafka/pull/14746#discussion_r1394496768
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -415,6 +439,27 @@ public ConsumerRecords<K, V> poll(final Duration timeout) {
return ConsumerRecords.empty();
} finally {
kafkaConsumerMetrics.recordPollEnd(timer.currentTimeMs());
+ wakeupTrigger.clearActiveTask();
Review Comment:
I think you are right @lucasbru - We need to ensure WakeupException will also be thrown when wakeup() is invoked. I wonder if we need to wrap the whole poll() in the future.
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #14746:
URL: https://github.com/apache/kafka/pull/14746#discussion_r1394571238
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -415,6 +439,27 @@ public ConsumerRecords<K, V> poll(final Duration timeout) {
return ConsumerRecords.empty();
} finally {
kafkaConsumerMetrics.recordPollEnd(timer.currentTimeMs());
+ wakeupTrigger.clearActiveTask();
+ }
+ }
+
+ private CompletableFuture<Void> setupWakeupTrigger() {
+ final CompletableFuture<Void> wakeupFuture = new CompletableFuture<>();
+ return wakeupTrigger.setActiveTask(wakeupFuture);
Review Comment:
I changed it to not use a future, but a little bit different from you proposal. The code became indeed simpler.
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #14746:
URL: https://github.com/apache/kafka/pull/14746#discussion_r1394572019
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -415,6 +439,27 @@ public ConsumerRecords<K, V> poll(final Duration timeout) {
return ConsumerRecords.empty();
} finally {
kafkaConsumerMetrics.recordPollEnd(timer.currentTimeMs());
+ wakeupTrigger.clearActiveTask();
Review Comment:
I removed it. We no longer need it.
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14746:
URL: https://github.com/apache/kafka/pull/14746#issuecomment-1813035808
hi @cadonna - I think we probably don't have to worry about it here. I have one comment: Is it possible to test poll would return normally after the consumer is woken up and invoke poll again?
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on PR #14746:
URL: https://github.com/apache/kafka/pull/14746#issuecomment-1813027684
I am still not sure what the conclusion is regarding: https://github.com/apache/kafka/pull/14746#issuecomment-1809158196
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14746:
URL: https://github.com/apache/kafka/pull/14746#issuecomment-1810851585
Hi @cadonna - When the consumer is woken up. The WakeupTrigger should complete the future exceptionally with WakeupException. To rethrow that exception during future.get(), you will need to examine the ExecutionException kind of like this:
```
catch (ExecutionException e) {
Throwable t = e.getCause();
if (t instanceof WakeupException)
throw new WakeupException();
else if (t instanceof KafkaException)
throw (KafkaException) t;
else
throw new KafkaException(t);
```
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on PR #14746:
URL: https://github.com/apache/kafka/pull/14746#issuecomment-1810259619
@kirktrue I am not sure, I understand your comment about `ExecutionException`. The future is not bound to any task, so can an `ExecutionException` ever be thrown?
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on PR #14746:
URL: https://github.com/apache/kafka/pull/14746#issuecomment-1814074041
> Is it possible to test poll would return normally after the consumer is woken up and invoke poll
That is a good idea!
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14746:
URL: https://github.com/apache/kafka/pull/14746#issuecomment-1815755117
Hi @cadonna - I was about to reply with the same idea you proposed. I think that would work. I wonder if we could just use a `BlockingQueue` for the fetchBuffer because fetchBuffer.poll(time) blocks until non-empty or timeout.
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on PR #14746:
URL: https://github.com/apache/kafka/pull/14746#issuecomment-1816006382
> Do you think it is worth adding a ticket about using the BackgroundEventQueue for the fetches? Let me know if you have a concrete idea about how to implement this.
Let me think about it.
--
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-15555: Ensure wakeups are handled correctly in poll() [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #14746:
URL: https://github.com/apache/kafka/pull/14746#discussion_r1397061812
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchBuffer.java:
##########
@@ -52,6 +53,8 @@ public class FetchBuffer implements AutoCloseable {
private final Condition notEmptyCondition;
private final IdempotentCloser idempotentCloser = new IdempotentCloser();
+ private final AtomicBoolean wakedUp = new AtomicBoolean(false);
Review Comment:
nit: wokenUp?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchBuffer.java:
##########
@@ -185,6 +188,16 @@ void awaitNotEmpty(Timer timer) {
}
}
+ void wakeup() {
Review Comment:
nit: `wakeup` vs. `wakeUp` capitalization
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchBuffer.java:
##########
@@ -166,7 +169,7 @@ void awaitNotEmpty(Timer timer) {
try {
lock.lock();
- while (isEmpty()) {
+ while (isEmpty() && !wakedUp.compareAndSet(true, false)) {
Review Comment:
Nice! I was first convinced that we have an interleaving here that misses the wakeup, but `wakedUp` takes care.
--
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