@Override
    public GroupAssignment assign(Cluster metadata, GroupSubscription groupSubscription) {
        Map<String, Subscription> subscriptions = groupSubscription.groupSubscription();
        Set<String> allSubscribedTopics = new HashSet<>();
        for (Map.Entry<String, Subscription> subscriptionEntry : subscriptions.entrySet())
            allSubscribedTopics.addAll(subscriptionEntry.getValue().topics());

        Map<String, List<PartitionInfo>> partitionsPerTopic = new HashMap<>();
        for (String topic : allSubscribedTopics) {
            List<PartitionInfo> partitions = metadata.partitionsForTopic(topic);
            if (partitions != null && !partitions.isEmpty()) {
                partitions = new ArrayList<>(partitions);
                partitions.sort(Comparator.comparingInt(PartitionInfo::partition));
                partitionsPerTopic.put(topic, partitions);
            } else {
                log.debug("Skipping assignment for topic {} since no metadata is available", topic);
            }
        }

        Map<String, List<TopicPartition>> rawAssignments = assignPartitions(partitionsPerTopic, subscriptions);

        // this class maintains no user data, so just wrap the results
        Map<String, Assignment> assignments = new HashMap<>();
        for (Map.Entry<String, List<TopicPartition>> assignmentEntry : rawAssignments.entrySet())
            assignments.put(assignmentEntry.getKey(), new Assignment(assignmentEntry.getValue()));
        return new GroupAssignment(assignments);
    }
Default implementation of assignPartitions() that does not include racks. This is only
included to avoid breaking any custom implementation that extends AbstractPartitionAssignor.
Note that this class is internal, but to be safe, we are maintaining compatibility.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractPartitionAssignor.java