@Override
    public GroupAssignment assign(
        GroupSpec groupSpec,
        SubscribedTopicDescriber subscribedTopicDescriber
    ) throws PartitionAssignorException {
        if (groupSpec.memberIds().isEmpty())
            return new GroupAssignment(Collections.emptyMap());

        if (groupSpec.subscriptionType().equals(HOMOGENEOUS)) {
            LOG.debug("Detected that all members are subscribed to the same set of topics, invoking the "
                + "homogeneous assignment algorithm");
            return new UniformHomogeneousAssignmentBuilder(groupSpec, subscribedTopicDescriber)
                .build();
        } else {
            LOG.debug("Detected that the members are subscribed to different sets of topics, invoking the "
                + "heterogeneous assignment algorithm");
            return new UniformHeterogeneousAssignmentBuilder(groupSpec, subscribedTopicDescriber)
                .build();
        }
    }
Perform the group assignment given the current members and
topics metadata.

@param groupSpec                     The assignment specification that included member metadata.
@param subscribedTopicDescriber      The topic and cluster metadata describer {@link SubscribedTopicDescriber}.
@return The new target assignment for the group.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/UniformAssignor.java