dongnuo123 commented on code in PR #16145:
URL: https://github.com/apache/kafka/pull/16145#discussion_r1622804714
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -806,7 +809,29 @@ public void validateOffsetCommit(
if (memberEpoch < 0 && members().isEmpty()) return;
final ConsumerGroupMember member = getOrMaybeCreateMember(memberId,
false);
- validateMemberEpoch(memberEpoch, member.memberEpoch());
+ if (member.useClassicProtocol()) {
+ validateMemberInstanceId(member, groupInstanceId);
+
+ try {
+ validateMemberEpoch(memberEpoch, member.memberEpoch());
+ } catch (StaleMemberEpochException ex) {
+ // StaleMemberEpochException is not supported in the classic
protocol. We throw
+ // IllegalGenerationException instead for compatibility.
+ throw new IllegalGenerationException(String.format("Invalid
offset commit because the "
+ + "received generation id %d does not match the expected
member epoch %d.",
+ memberEpoch, member.memberEpoch()));
+ }
+
+ if (member.memberEpoch() < groupEpoch() ||
+ member.state() == MemberState.UNREVOKED_PARTITIONS ||
+ (member.state() == MemberState.UNRELEASED_PARTITIONS &&
!waitingOnUnreleasedPartition(member))) {
+ throw new RebalanceInProgressException(String.format("Invalid
offset commit because" +
+ " a new rebalance has been triggered in group %s and
member %s should rejoin to catch up.",
+ groupId(), member.memberId()));
+ }
Review Comment:
Got it, now I know why the classic consumer offset commit validation doesn't
check PREPARING_REBALANCE.
--
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]
