@Override
    public GroupAssignment assign(
        GroupSpec groupSpec,
        SubscribedTopicDescriber subscribedTopicDescriber
    ) throws PartitionAssignorException {
        if (groupSpec.memberIds().isEmpty()) {
            return new GroupAssignment(Collections.emptyMap());
        } else if (groupSpec.subscriptionType() == SubscriptionType.HOMOGENEOUS) {
            return assignHomogeneousGroup(groupSpec, subscribedTopicDescriber);
        } else {
            return assignHeterogeneousGroup(groupSpec, subscribedTopicDescriber);
        }
    }
Assigns partitions to members based on their topic subscriptions and the properties of a range assignor:

@param groupSpec                     The group specification containing the member information.
@param subscribedTopicDescriber      The describer for subscribed topics to get the number of partitions.
@return The group's assignment with the partition assignments for each member.
@throws PartitionAssignorException if any member is subscribed to a non-existent topic.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java