@Override
    public Map<String, List<TopicPartition>> assignPartitions(Map<String, List<PartitionInfo>> partitionsPerTopic,
                                                              Map<String, Subscription> subscriptions) {
        Map<String, List<MemberInfo>> consumersPerTopic = consumersPerTopic(subscriptions);
        Map<String, String> consumerRacks = consumerRacks(subscriptions);
        List<TopicAssignmentState> topicAssignmentStates = partitionsPerTopic.entrySet().stream()
                .filter(e -> !e.getValue().isEmpty())
                .map(e -> new TopicAssignmentState(e.getKey(), e.getValue(), consumersPerTopic.get(e.getKey()), consumerRacks))
                .collect(Collectors.toList());

        Map<String, List<TopicPartition>> assignment = new HashMap<>();
        subscriptions.keySet().forEach(memberId -> assignment.put(memberId, new ArrayList<>()));

        boolean useRackAware = topicAssignmentStates.stream().anyMatch(t -> t.needsRackAwareAssignment);
        if (useRackAware)
            assignWithRackMatching(topicAssignmentStates, assignment);

        topicAssignmentStates.forEach(t -> assignRanges(t, (c, tp) -> true, assignment));

        if (useRackAware)
            assignment.values().forEach(list -> list.sort(PARTITION_COMPARATOR));
        return 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