lianetm commented on code in PR #20521:
URL: https://github.com/apache/kafka/pull/20521#discussion_r2411728784
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -726,6 +711,71 @@ private void process(final
StreamsOnAllTasksLostCallbackCompletedEvent event) {
requestManagers.streamsMembershipManager.get().onAllTasksLostCallbackCompleted(event);
}
+ private void process(final AsyncPollEvent event) {
+ log.trace("Processing poll logic for {}", event);
+
+ // Trigger a reconciliation that can safely commit offsets if needed
to rebalance,
+ // as we're processing before any new fetching starts
+
requestManagers.consumerMembershipManager.ifPresent(consumerMembershipManager ->
+ consumerMembershipManager.maybeReconcile(true));
+
+ if (requestManagers.commitRequestManager.isPresent()) {
+ CommitRequestManager commitRequestManager =
requestManagers.commitRequestManager.get();
+ commitRequestManager.updateTimerAndMaybeCommit(event.pollTimeMs());
+
+ requestManagers.consumerHeartbeatRequestManager.ifPresent(hrm -> {
+ hrm.membershipManager().onConsumerPoll();
+ hrm.resetPollTimer(event.pollTimeMs());
+ });
+ requestManagers.streamsGroupHeartbeatRequestManager.ifPresent(hrm
-> {
+ hrm.membershipManager().onConsumerPoll();
+ hrm.resetPollTimer(event.pollTimeMs());
+ });
+ }
+
+ log.trace("Processing check and update positions logic for {}", event);
Review Comment:
yeap, but what's the value of this 2nd one? The event is the same and the
logic in-between the 2 logs is pretty simple, no early returns, just triggering
actions or updating vars, I expect in practice we'll always have 2 identical
lines right? (potentially on every poll iteration)
--
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]
