private void assignRanges(TopicAssignmentState assignmentState,
                              BiFunction<String, TopicPartition, Boolean> mayAssign,
                              Map<String, List<TopicPartition>> assignment) {
        for (String consumer : assignmentState.consumers.keySet()) {
            if (assignmentState.unassignedPartitions.isEmpty())
                break;
            List<TopicPartition> assignablePartitions = assignmentState.unassignedPartitions.stream()
                    .filter(tp -> mayAssign.apply(consumer, tp))
                    .limit(assignmentState.maxAssignable(consumer))
                    .collect(Collectors.toList());
            if (assignablePartitions.isEmpty())
                continue;

            assign(consumer, assignablePartitions, assignmentState, assignment);
        }
    }
Performs range assignment of the specified partitions for the consumers with the provided subscriptions.
If rack-awareness is enabled for one or more consumers, we perform rack-aware assignment first to assign
the subset of partitions that can be aligned on racks, while retaining the same co-partitioning and
per-topic balancing guarantees as non-rack-aware range assignment. The remaining partitions are assigned
using standard non-rack-aware range assignment logic, which may result in mis-aligned racks.
clients/src/main/java/org/apache/kafka/clients/consumer/RangeAssignor.java