lianetm commented on code in PR #18737:
URL: https://github.com/apache/kafka/pull/18737#discussion_r1958555476
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -748,9 +748,14 @@ public ConsumerRecords<K, V> poll(final Duration timeout) {
}
do {
-
+ PollEvent event = new PollEvent(timer.currentTimeMs());
// Make sure to let the background thread know that we are
still polling.
- applicationEventHandler.add(new
PollEvent(timer.currentTimeMs()));
+ // This will trigger async auto-commits of consumed positions
when hitting
+ // the interval time or reconciling new assignments
+ applicationEventHandler.add(event);
+ // Wait for reconciliation and auto-commit to be triggered, to
ensure all commit requests
+ // retrieve the positions to commit before proceeding with
fetching new records
+ ConsumerUtils.getResult(event.reconcileAndAutoCommit());
Review Comment:
should we pass the default api timeout here? Under normal execution, this
will just complete right away (local actions in the background), but if the
background thread is faulty (ie. died) and the event can't be processed, the
consumer would hang here indefinitely (instead of timing out). Note that I
suggest the default api timeout and not the timeout from param because we could
have poll(ZERO), and that 0 shouldn't apply to the inter-thread communication,
which is what we're doing here. Same for the blocking call we added for
offsetsReady.
We do this same approach in other api calls btw, ex. seek. 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
