junrao commented on code in PR #18737:
URL: https://github.com/apache/kafka/pull/18737#discussion_r1957375625
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -206,13 +206,25 @@ public void process(ApplicationEvent event) {
}
private void process(final PollEvent event) {
+ // Trigger a reconciliation that can safely commit offsets if needed
to rebalance,
+ // as we're processing before any new fetching starts in the app thread
+
requestManagers.consumerMembershipManager.ifPresent(consumerMembershipManager ->
+ consumerMembershipManager.maybeReconcile(true));
if (requestManagers.commitRequestManager.isPresent()) {
- requestManagers.commitRequestManager.ifPresent(m ->
m.updateAutoCommitTimer(event.pollTimeMs()));
+ CommitRequestManager commitRequestManager =
requestManagers.commitRequestManager.get();
+ commitRequestManager.updateTimerAndMaybeCommit(event.pollTimeMs());
+ // all commit request generation points have been passed,
+ // so it's safe to notify the app thread could proceed and start
fetching
+ event.future().complete(null);
Review Comment:
Technically, the event hasn't been fully processed yet. So, it's a bit weird
to complete the event midway. Perhaps we could introduce a separate future like
`reconcileAndAutoCommit` and wait for that.
--
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
