squah-confluent commented on code in PR #20000:
URL: https://github.com/apache/kafka/pull/20000#discussion_r2582004284
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/UniformHomogeneousAssignmentBuilder.java:
##########
@@ -240,13 +277,75 @@ private void maybeRevokePartitions() {
}
}
+ /**
+ * Assign the unassigned partitions to the unfilled members if member and
partition racks are matched.
+ */
+ private void assignRackAwarenessRemainingPartitions() {
+ // Assign partitions to members with descending order. This avoids the
cost of shifting elements.
+ for (int i = unassignedPartitions.size() - 1; i >= 0; i--) {
+ TopicIdPartition tip = unassignedPartitions.get(i);
+ boolean isPartitionAssigned = false;
+
+ for (var unfilledMembersIter = unfilledMembers.iterator();
unfilledMembersIter.hasNext(); ) {
+ MemberWithRemainingQuota unfilledMember =
unfilledMembersIter.next();
+ if (unfilledMember.remainingQuota() == 0 &&
remainingMembersToGetAnExtraPartition == 0) {
+ unfilledMembersIter.remove();
+ continue;
+ }
+
+ String memberId = unfilledMember.memberId;
+ if
(!AssignorHelpers.isRackMatch(groupSpec.memberSubscription(memberId).rackId(),
+ partitionRacks.getOrDefault(tip, Set.of()))) {
+ continue;
+ }
+
+ Map<Uuid, Set<Integer>> newAssignment =
targetAssignment.get(memberId).partitions();
+ if (AssignorHelpers.isImmutableMap(newAssignment)) {
+ // If the new assignment is immutable, we must create a
deep copy of it
+ // before altering it.
+ newAssignment =
AssignorHelpers.deepCopyAssignment(newAssignment);
+ targetAssignment.put(memberId, new
MemberAssignmentImpl(newAssignment));
Review Comment:
https://github.com/apache/kafka/pull/20097 introduced a performance
regression. After that change `MemberAssignmentImpl` always wraps the new
assignment in another immutable map, so we will always deep copy here. Can we
fix the regression in a separate PR?
--
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]
