squah-confluent commented on code in PR #20000:
URL: https://github.com/apache/kafka/pull/20000#discussion_r2582160658
##########
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:
I filed https://issues.apache.org/jira/browse/KAFKA-19955 and opened
https://github.com/apache/kafka/pull/21058 to fix it. It might improve the
rack-aware times by a bit.
--
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]
