jeffkbkim commented on code in PR #15974:
URL: https://github.com/apache/kafka/pull/15974#discussion_r1607031651
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/GroupSpecImpl.java:
##########
@@ -33,13 +35,21 @@ public class AssignmentSpec {
*/
private final SubscriptionType subscriptionType;
- public AssignmentSpec(
+ /**
+ * Reverse lookup map representing partitions per topic and
+ * their member assignment.
+ */
+ Map<Uuid, Map<Integer, String>> partitionAssignments;
Review Comment:
should this also be private final, and should the user be able to modify the
contents of the map?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -516,16 +522,78 @@ public Assignment targetAssignment(String memberId) {
return targetAssignment.getOrDefault(memberId, Assignment.EMPTY);
}
+ /**
+ * @return An immutable map containing all the topic partitions
+ * with their current member assignments.
+ */
+ public Map<Uuid, Map<Integer, String>> partitionAssignments() {
+ return Collections.unmodifiableMap(partitionAssignments);
+ }
+
/**
* Updates target assignment of a member.
*
* @param memberId The member id.
* @param newTargetAssignment The new target assignment.
*/
public void updateTargetAssignment(String memberId, Assignment
newTargetAssignment) {
+ updatePartitionAssignments(
+ memberId,
+ targetAssignment.getOrDefault(memberId, new
Assignment(Collections.emptyMap())),
+ newTargetAssignment
+ );
targetAssignment.put(memberId, newTargetAssignment);
}
+ /**
+ * Updates partition assignments of the topics.
+ *
+ * @param memberId The member Id.
+ * @param oldTargetAssignment The old target assignment.
+ * @param newTargetAssignment The new target assignment.
+ *
+ * Package private for testing.
+ */
+ void updatePartitionAssignments(
+ String memberId,
+ Assignment oldTargetAssignment,
+ Assignment newTargetAssignment
+ ) {
+ // Combine keys from both old and new assignments.
+ Set<Uuid> allTopicIds = new HashSet<>();
+ allTopicIds.addAll(oldTargetAssignment.partitions().keySet());
+ allTopicIds.addAll(newTargetAssignment.partitions().keySet());
+
+ for (Uuid topicId : allTopicIds) {
+ Set<Integer> oldPartitions =
oldTargetAssignment.partitions().getOrDefault(topicId, Collections.emptySet());
+ Set<Integer> newPartitions =
newTargetAssignment.partitions().getOrDefault(topicId, Collections.emptySet());
+
+ TimelineHashMap<Integer, String> topicPartitionAssignment =
partitionAssignments.computeIfAbsent(
+ topicId, k -> new TimelineHashMap<>(snapshotRegistry,
Math.max(oldPartitions.size(), newPartitions.size()))
+ );
+
+ // Remove partitions that aren't present in the new assignment.
+ for (Integer partition : oldPartitions) {
+ if (!newPartitions.contains(partition) &&
memberId.equals(topicPartitionAssignment.get(partition))) {
Review Comment:
on `memberId.equals(topicPartitionAssignment.get(partition))`: will this
ever be false? it seems we would always assign a partition to a member only
when it's been removed by the previous member, 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
