kirktrue commented on code in PR #20521:
URL: https://github.com/apache/kafka/pull/20521#discussion_r2409174817
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -726,6 +730,96 @@ 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);
+ CompletableFuture<Boolean> updatePositionsFuture =
requestManagers.offsetsRequestManager.updateFetchPositions(event.deadlineMs());
+
+ // To maintain the flow from ClassicKafkaConsumer, the
check-and-update-positions logic should be allowed
+ // to time out before moving on to the logic for sending fetch
requests. This is achieved by using the event
+ // reaper and allowing it to expire the check-and-update-positions
future.
+ applicationEventReaper.ifPresent(reaper -> {
+ CompletableEvent<Boolean> pseudoEvent = new CompletableEvent<>() {
+ @Override
+ public CompletableFuture<Boolean> future() {
+ return updatePositionsFuture;
+ }
+
+ @Override
+ public long deadlineMs() {
+ return event.deadlineMs();
+ }
+
+ @Override
+ public String toString() {
+ return getClass().getSimpleName() +
"{updatePositionsFuture=" + updatePositionsFuture + ", deadlineMs=" +
event.deadlineMs() + '}';
+ }
+ };
+
+ reaper.add(pseudoEvent);
+ });
Review Comment:
I've removed the timeout for the check and update positions 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
