class RangeAssignor extends AbstractPartitionAssignor {

    @Override
    public Map<String/* memberId */, List<TopicPartition>> assign(Map<String/* topic */, Integer> partitionsPerTopic,
        Map<String/* memberId */, List<String/* topic */>> subscriptions) {

        // 将原本的 member订阅了哪几个topic，转换为topic 被哪几个member订阅了 topicPerConsumer => consumerPerTopic
        Map<String/* topic */, List<String/* memberId */>> consumersPerTopic = consumersPerTopic(subscriptions);

        // 为每个 member new一个空的TopicPartition订阅集合
        Map<String/* memberId */, List<TopicPartition>> assignment = new HashMap<>();
        for (String memberId : subscriptions.keySet())
            assignment.put(memberId, new ArrayList<TopicPartition>());

        // 开始分配
        for (Map.Entry<String, List<String>> topicEntry : consumersPerTopic.entrySet()) {
            String topic = topicEntry.getKey();
            List<String/* memberId */> consumersForTopic = topicEntry.getValue();

            // 根据topic名字从 partitionsPerTopic 到partition下有多少个topic
            Integer numPartitionsForTopic = partitionsPerTopic.get(topic);
            if (numPartitionsForTopic == null) {
                continue;
            }

            Collections.sort(consumersForTopic);

            int numConsumersForConsumer = consumersForTopic.size();

            int N_numPartitionsPerConsumer = numPartitionsForTopic / numConsumersForConsumer;
            int M_consumersWithExtraPartition = numPartitionsForTopic % numConsumersForConsumer;

            List<TopicPartition> partitions = AbstractPartitionAssignor.partitions(topic, numPartitionsForTopic);

            // n = 分区数/消费者，m = 分区数%消费者，前m个消费者分配n+1个分区（每有一个余数，分配多一个），
            // 后面的消费者消费n个分区
            for (int i = 0, n = consumersForTopic.size(); i < n; i++) {
                int start = N_numPartitionsPerConsumer * i + Math.min(i, M_consumersWithExtraPartition);
                int length = N_numPartitionsPerConsumer + (i + 1 > M_consumersWithExtraPartition ? 0 : 1);
                assignment.get(consumersForTopic.get(i))
                          .addAll(partitions.subList(start, start + length));
            }
        }
        return assignment;
    }

}