lucliu1108 commented on code in PR #21558:
URL: https://github.com/apache/kafka/pull/21558#discussion_r2884659960
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Utils.java:
##########
@@ -201,17 +231,68 @@ public static
List<ConsumerGroupHeartbeatRequestData.TopicPartitions> toTopicPar
}
/**
- * Creates a map of topic id and partition set from a list of consumer
group TopicPartitions.
+ * Creates a map of topic id and partition with assignment epochs from a
list of consumer group TopicPartitions.
*
- * @param topicPartitionsList The list of TopicPartitions.
- * @return a map of topic id and partition set.
+ * @param topicPartitions The list of TopicPartitions.
+ * @param defaultEpoch The default epoch to use when the epoch information
is not available for a partition.
+ * @return a map of topic id and partitions with assignment epochs.
*/
- public static Map<Uuid, Set<Integer>> assignmentFromTopicPartitions(
- List<ConsumerGroupCurrentMemberAssignmentValue.TopicPartitions>
topicPartitionsList
+ public static Map<Uuid, Map<Integer, Integer>>
assignmentFromTopicPartitions(
+ List<ConsumerGroupCurrentMemberAssignmentValue.TopicPartitions>
topicPartitions,
+ int defaultEpoch
) {
- return topicPartitionsList.stream().collect(Collectors.toMap(
- ConsumerGroupCurrentMemberAssignmentValue.TopicPartitions::topicId,
- topicPartitions -> Collections.unmodifiableSet(new
HashSet<>(topicPartitions.partitions()))));
+ // For legacy static member, the defaultEpoch could be -2
(LEAVE_GROUP_STATIC_MEMBER_EPOCH).
+ // But we want to ensure the default memberEpoch assigned is
non-negative.
+ int adjustedDefaultEpoch = Math.max(defaultEpoch, 0);
+ Map<Uuid, Map<Integer, Integer>> assignmentWithEpochs = new
HashMap<>();
+ for (ConsumerGroupCurrentMemberAssignmentValue.TopicPartitions tp :
topicPartitions) {
+ Map<Integer, Integer> partitionEpochs = new HashMap<>();
+ List<Integer> partitions = tp.partitions();
+ List<Integer> epochs = tp.assignmentEpochs();
+ if (epochs != null && !epochs.isEmpty() && epochs.size() !=
partitions.size()) {
Review Comment:
If we don't allow for not-null empty epoch array, for legacy record that
doesn't have `assignmentEpochs` tagged field before, will the
`assignmentEpochs` remain empty list instead of null?
For streamsGroup equivalent, in `TasksTupleWIthEpochs.java` 's
[`parseActiveTasksWithEpochs`](https://github.com/apache/kafka/blob/trunk/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TasksTupleWithEpochs.java#L140-L157),
it treats empty list same as null. Do we want to do the same here for
consumer group?
--
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]
