private void assignRackAwareRoundRobin(List<TopicPartition> unassignedPartitions) {
            if (rackInfo.consumerRacks.isEmpty())
                return;
            int nextUnfilledConsumerIndex = 0;
            Iterator<TopicPartition> unassignedIter = unassignedPartitions.iterator();
            while (unassignedIter.hasNext()) {
                TopicPartition unassignedPartition = unassignedIter.next();
                String consumer = null;
                int nextIndex = rackInfo.nextRackConsumer(unassignedPartition, unfilledMembersWithUnderMinQuotaPartitions, nextUnfilledConsumerIndex);
                if (nextIndex >= 0) {
                    consumer = unfilledMembersWithUnderMinQuotaPartitions.get(nextIndex);
                    int assignmentCount = assignment.get(consumer).size() + 1;
                    if (assignmentCount >= minQuota) {
                        unfilledMembersWithUnderMinQuotaPartitions.remove(consumer);
                        // Only add this consumer if the current num members at maxQuota is less than the expected number
                        // since a consumer at minQuota can only be considered unfilled if it's possible to add another partition,
                        // which would bump it to maxQuota and exceed the expectedNumMembersWithOverMinQuotaPartitions
                        if (assignmentCount < maxQuota && (currentNumMembersWithOverMinQuotaPartitions < expectedNumMembersWithOverMinQuotaPartitions)) {
                            unfilledMembersWithExactlyMinQuotaPartitions.add(consumer);
                        }
                    } else {
                        nextIndex++;
                    }
                    nextUnfilledConsumerIndex = unfilledMembersWithUnderMinQuotaPartitions.isEmpty() ? 0 : nextIndex % unfilledMembersWithUnderMinQuotaPartitions.size();
                } else if (!unfilledMembersWithExactlyMinQuotaPartitions.isEmpty()) {
                    int firstIndex = rackInfo.nextRackConsumer(unassignedPartition, unfilledMembersWithExactlyMinQuotaPartitions, 0);
                    if (firstIndex >= 0) {
                        consumer = unfilledMembersWithExactlyMinQuotaPartitions.get(firstIndex);
                        if (assignment.get(consumer).size() + 1 == maxQuota) {
                            unfilledMembersWithExactlyMinQuotaPartitions.remove(firstIndex);
                            currentNumMembersWithOverMinQuotaPartitions++;
                            // Clear this once the current num consumers over minQuota reaches the expected number since this
                            // means all consumers at minQuota are now considered filled
                            if (currentNumMembersWithOverMinQuotaPartitions == expectedNumMembersWithOverMinQuotaPartitions) {
                                unfilledMembersWithExactlyMinQuotaPartitions.clear();
                            }
                        }
                    }
                }

                if (consumer != null) {
                    assignNewPartition(unassignedPartition, consumer);
                    unassignedIter.remove();
                }
            }
        }
Constructs a constrained assignment builder.

@param partitionsPerTopic                   The partitions for each subscribed topic
@param rackInfo                             Rack information for consumers and racks
@param consumerToOwnedPartitions            Each consumer's previously owned and still-subscribed partitions
@param partitionsWithMultiplePreviousOwners The partitions being claimed in the previous assignment of multiple consumers
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java