class KafkaUtils {

    public static List<Partition> calculatePartitionsForTask(GlobalPartitionInformation partitionInformation, int totalTasks, int taskIndex) {
        Preconditions.checkArgument(taskIndex < totalTasks, "task index must be less that total tasks");
        List<Partition> partitions = partitionInformation.getOrderedPartitions();
        //kafka的日志分区数和spout的任务数.
        int numPartitions = partitions.size();
        //如果spout的任务数比日志分区数还要多, 说明有些task会空闲. 一般要一个任务处理多个日志分区, 即totalTasks <= partitions
        if (numPartitions < totalTasks) {
            LOG.warn("there are more tasks than partitions (tasks: " + totalTasks + "; partitions: " + numPartitions + "), some tasks will be idle");
        }

        List<Partition> taskPartitions = new ArrayList<Partition>();
        //假设有10个分区日志,有5个任务. taskIndex=0, 分配日志分区=0,5
        //taskIndex=1, 分配日志分区=1,6. 实际上是顺序为task分配一个日志分区: 每个task分配一个顺序的日志分区
        //如果还有剩余的日志分区没有指定task,再从头开始为所有的task继续顺序分配
        for (int i = taskIndex; i < numPartitions; i += totalTasks) {
            Partition taskPartition = partitions.get(i);
            taskPartitions.add(taskPartition);
        }
        logPartitionMapping(totalTasks, taskIndex, taskPartitions);
        return taskPartitions;
    }

}