lianetm commented on code in PR #14364:
URL: https://github.com/apache/kafka/pull/14364#discussion_r1331828912
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -146,8 +152,9 @@ private void onFailure(final Throwable exception, final
long responseTimeMs) {
logger.debug("failed sending heartbeat due to {}",
exception.getMessage());
}
- private void onSuccess(final ConsumerGroupHeartbeatResponse response, long
currentTimeMs) {
+ private void onResponse(final ConsumerGroupHeartbeatResponse response,
long currentTimeMs) {
Review Comment:
Again brainstorming based on the
[draft](https://github.com/apache/kafka/pull/14413), this would be much
simplified with the move of the error handling more into the MembershipManager
I expect. Here, instead of having to paths to update state (now there are 2
calls, one to membershipManager.updateState and another for all the error
handling), we could simply have something like:
```
private void onResponse(final ConsumerGroupHeartbeatResponse response, long
currentTimeMs) {
if (response.data().errorCode() == Errors.NONE.code()) {
// Heartbeat manager specifics for success - nothing affecting
state
} else {
// Heartbeat manager specifics for failure - nothing affecting
state & no error handling other than identifying when
coordinatorRequestManager.markCoordinatorUnknown is needed, which should be
here I think
}
}
// Update state - single point of interaction with the membershipMgr
Optional<Errors> error =
membershipManager.updateState(response.data());
if (error.isPresent()) {
nonRetriableErrorHandler.handle(error.get().exception());
}
}
```
--
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
