dongnuo123 commented on code in PR #15268:
URL: https://github.com/apache/kafka/pull/15268#discussion_r1467807980
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3467,6 +3476,358 @@ public void maybeDeleteGroup(String groupId,
List<Record> records) {
}
}
+ private JoinGroupRequestProtocol throwIfProtocolUnmatched(
+ ConsumerGroupMember member,
+ JoinGroupRequestData.JoinGroupRequestProtocolCollection protocols
+ ) {
+ for (JoinGroupRequestData.JoinGroupRequestProtocol protocol :
protocols) {
+ final ByteBuffer buffer = ByteBuffer.wrap(protocol.metadata());
+ ConsumerProtocol.deserializeVersion(buffer);
+ final Optional<Integer> generationId =
ConsumerProtocol.deserializeSubscription(buffer, (short) 0).generationId();
+
+ // If the generation id is provided, it must match the member
epoch.
+ if (!generationId.isPresent() || generationId.get() ==
member.memberEpoch()) {
+ // TODO: need a list of all available server assignors
+ if
(UniformAssignor.UNIFORM_ASSIGNOR_NAME.equals(protocol.name())
+ ||
RangeAssignor.RANGE_ASSIGNOR_NAME.equals(protocol.name())) {
+ return protocol;
+ }
+ }
+ }
+ throw new FencedMemberEpochException("The JoinGroup request doesn't
have a matched generation id from a " +
+ "protocol supported by the server assignors with the epoch of the
member known by the group coordinator (" +
+ member.memberEpoch() + ").");
Review Comment:
maybe handle the case where generation id doesn't match the member epoch.
--
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
