lianetm commented on code in PR #18050:
URL: https://github.com/apache/kafka/pull/18050#discussion_r1904255768
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -196,6 +198,14 @@ public NetworkClientDelegate.PollResult poll(final long
currentTimeMs) {
return new NetworkClientDelegate.PollResult(timeUntilNextPoll,
requests);
}
+ private void maybeFailPendingRequestsOnCoordinatorFatalError() {
+ Optional<Throwable> fatalError =
coordinatorRequestManager.fatalError();
+ if (fatalError.isPresent()) {
+ pendingRequests.unsentOffsetCommits.forEach(request ->
request.future.completeExceptionally(fatalError.get()));
+ pendingRequests.unsentOffsetFetches.forEach(request ->
request.future.completeExceptionally(fatalError.get()));
Review Comment:
don't we need to remove the pending requests that we're failing here?
`pendingRequests` are usually cleared on poll
[clearAll](https://github.com/apache/kafka/blob/64bbdb1a0319044d1fa60e3836ab4052c7a28c12/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java#L1221),
but if there is a fatal error we won't make it there right?
If this fix makes sense, let's try to add a unit test to ensure that we
don't leave any pending commit request that fails after a fatal error with the
coordinator.
--
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
