m1a2st commented on code in PR #18050:
URL: https://github.com/apache/kafka/pull/18050#discussion_r1885087039
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -176,9 +176,15 @@ public CommitRequestManager(
*/
@Override
public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
- // poll only when the coordinator node is known.
- if (coordinatorRequestManager.coordinator().isEmpty())
+ // poll when the coordinator node is known and fatal errors are not
present
+ if (coordinatorRequestManager.coordinator().isEmpty()) {
+ 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()));
+ }
return EMPTY;
Review Comment:
> The CommitRequestManager does not clear the fatal error, but the
AbstractHeartbeatRequestManager does. Can you explain and/or add some comments
so that others can understand the distinction between these two cases?
I addressed some comments on this, If you have other question, Please let me
know.
--
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]
