private void assignRoundRobin(List<TopicPartition> unassignedPartitions) {

            Iterator<String> unfilledConsumerIter = unfilledMembersWithUnderMinQuotaPartitions.iterator();
            // Round-Robin filling remaining members up to the expected numbers of maxQuota, otherwise, to minQuota
            for (TopicPartition unassignedPartition : unassignedPartitions) {
                String consumer;
                if (unfilledConsumerIter.hasNext()) {
                    consumer = unfilledConsumerIter.next();
                } else {
                    if (unfilledMembersWithUnderMinQuotaPartitions.isEmpty() && unfilledMembersWithExactlyMinQuotaPartitions.isEmpty()) {
                        // Should not enter here since we have calculated the exact number to assign to each consumer.
                        // This indicates issues in the assignment algorithm
                        int currentPartitionIndex = unassignedPartitions.indexOf(unassignedPartition);
                        log.error("No more unfilled consumers to be assigned. The remaining unassigned partitions are: {}",
                                unassignedPartitions.subList(currentPartitionIndex, unassignedPartitions.size()));
                        throw new IllegalStateException("No more unfilled consumers to be assigned.");
                    } else if (unfilledMembersWithUnderMinQuotaPartitions.isEmpty()) {
                        consumer = unfilledMembersWithExactlyMinQuotaPartitions.poll();
                    } else {
                        unfilledConsumerIter = unfilledMembersWithUnderMinQuotaPartitions.iterator();
                        consumer = unfilledConsumerIter.next();
                    }
                }

                int currentAssignedCount = assignNewPartition(unassignedPartition, consumer);

                if (currentAssignedCount == minQuota) {
                    unfilledConsumerIter.remove();
                    unfilledMembersWithExactlyMinQuotaPartitions.add(consumer);
                } else if (currentAssignedCount == maxQuota) {
                    currentNumMembersWithOverMinQuotaPartitions++;
                    if (currentNumMembersWithOverMinQuotaPartitions == expectedNumMembersWithOverMinQuotaPartitions) {
                        // We only start to iterate over the "potentially unfilled" members at minQuota after we've filled
                        // all members up to at least minQuota, so once the last minQuota member reaches maxQuota, we
                        // should be done. But in case of some algorithmic error, just log a warning and continue to
                        // assign any remaining partitions within the assignment constraints
                        if (unassignedPartitions.indexOf(unassignedPartition) != unassignedPartitions.size() - 1) {
                            log.error("Filled the last member up to maxQuota but still had partitions remaining to assign, "
                                    + "will continue but this indicates a bug in the assignment.");
                        }
                    }
                }
            }
        }
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