philipnee commented on code in PR #14364:
URL: https://github.com/apache/kafka/pull/14364#discussion_r1336389075
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -115,25 +115,37 @@ public int memberEpoch() {
@Override
public void updateState(ConsumerGroupHeartbeatResponseData response) {
- if (response.errorCode() == Errors.NONE.code()) {
- this.memberId = response.memberId();
- this.memberEpoch = response.memberEpoch();
- ConsumerGroupHeartbeatResponseData.Assignment assignment =
response.assignment();
- if (assignment != null) {
- setTargetAssignment(assignment);
- }
- maybeTransitionToStable();
- } else {
- if (response.errorCode() == Errors.FENCED_MEMBER_EPOCH.code() ||
response.errorCode() == Errors.UNKNOWN_MEMBER_ID.code()) {
- resetEpoch();
- transitionTo(MemberState.FENCED);
- } else if (response.errorCode() ==
Errors.UNRELEASED_INSTANCE_ID.code()) {
- transitionTo(MemberState.FAILED);
- }
- // TODO: handle other errors here to update state accordingly,
mainly making the
- // distinction between the recoverable errors and the fatal ones,
that should FAILED
- // the member
+ if (response.errorCode() != Errors.NONE.code()) {
+ String errorMessage = String.format(
+ "Unexpected error in Heartbeat response. Expected no
error, but received: %s",
+ Errors.forCode(response.errorCode())
+ );
+ throw new IllegalStateException(errorMessage);
+ }
+ this.memberId = response.memberId();
+ this.memberEpoch = response.memberEpoch();
+ ConsumerGroupHeartbeatResponseData.Assignment assignment =
response.assignment();
+ if (assignment != null) {
+ setTargetAssignment(assignment);
}
+ maybeTransitionToStable();
+ }
+
+ @Override
+ public void fenceMember() {
+ resetEpoch();
+ transitionTo(MemberState.FENCED);
+ }
+
+ @Override
+ public void transitionToFailure() {
+ transitionTo(MemberState.FAILED);
+ }
+
+ @Override
+ public boolean shouldSendHeartbeat() {
Review Comment:
i think you are right.
--
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
