dajac commented on code in PR #17853:
URL: https://github.com/apache/kafka/pull/17853#discussion_r1847932090
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -939,12 +941,20 @@ public static ConsumerGroup fromClassicGroup(
consumerGroup.setTargetAssignmentEpoch(classicGroup.generationId());
classicGroup.allMembers().forEach(classicGroupMember -> {
- Map<Uuid, Set<Integer>> assignedPartitions = toTopicPartitionMap(
- ConsumerProtocol.deserializeConsumerProtocolAssignment(
- ByteBuffer.wrap(classicGroupMember.assignment())
- ),
- topicsImage
- );
+ // The assigned partition can be empty if the member just joined
and has never synced.
+ // We should accept the empty assignment.
+ Map<Uuid, Set<Integer>> assignedPartitions;
+ if (Arrays.equals(classicGroupMember.assignment(),
EMPTY_ASSIGNMENT)) {
+ assignedPartitions = Collections.emptyMap();
+ } else {
+ assignedPartitions = toTopicPartitionMap(
+ ConsumerProtocol.deserializeConsumerProtocolAssignment(
+ ByteBuffer.wrap(classicGroupMember.assignment())
+ ),
+ topicsImage
+ );
+ }
+
ConsumerProtocolSubscription subscription =
ConsumerProtocol.deserializeConsumerProtocolSubscription(
ByteBuffer.wrap(classicGroupMember.metadata(classicGroup.protocolName().get()))
Review Comment:
For my understanding. I think that we don't need a similar fix here because
every member is guaranteed to have metadata set when it joins. However, the
metadata could be malformed. In this case, the deserialization fails. I suppose
that it results in a unknown server error returned to the client. Is my
understanding correct?
--
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]
