private void assignCoPartitionedWithRackMatching(LinkedHashMap<String, Optional<String>> consumers,
                                                     int numPartitions,
                                                     Collection<TopicAssignmentState> assignmentStates,
                                                     Map<String, List<TopicPartition>> assignment) {

        Set<String> remainingConsumers = new LinkedHashSet<>(consumers.keySet());
        for (int i = 0; i < numPartitions; i++) {
            int p = i;

            Optional<String> matchingConsumer = remainingConsumers.stream()
                    .filter(c -> assignmentStates.stream().allMatch(t -> t.racksMatch(c, new TopicPartition(t.topic, p)) && t.maxAssignable(c) > 0))
                    .findFirst();
            if (matchingConsumer.isPresent()) {
                String consumer = matchingConsumer.get();
                assignmentStates.forEach(t -> assign(consumer, Collections.singletonList(new TopicPartition(t.topic, p)), t, assignment));

                if (assignmentStates.stream().noneMatch(t -> t.maxAssignable(consumer) > 0)) {
                    remainingConsumers.remove(consumer);
                    if (remainingConsumers.isEmpty())
                        break;
                }
            }
        }
    }
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