lianetm commented on code in PR #16844:
URL: https://github.com/apache/kafka/pull/16844#discussion_r1731586009
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -450,6 +452,12 @@ private void handleFatalFailure(Throwable error) {
membershipManager.transitionToFatal();
}
+ private void handleCoordinatorDisconnect(Throwable exception, long
currentTimeMs) {
+ if (exception instanceof DisconnectException) {
+
coordinatorRequestManager.markCoordinatorUnknown(exception.getMessage(),
currentTimeMs);
+ }
+ }
Review Comment:
With this we end up having the exact same logic to
`handleCoordinatorDisconnect` repeated on the managers that need it. What about
we encapsulate it in the `CoordinatorRequestManager`? Seems sensible for it to
be the component who know how to handle the error (and we only update there if
ever needed). So we would end up with the managers just calling:
`coordinatorRequestManager.handleCoordinatorDisconnect(exception,
currentTimeMs)`. What do you think?
--
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]
