Flink 源码分析之 FlinkConsumer 是如何保证一个 partition 对应一个 thread 的

//获取fixed topic's or topic pattern 's   partitions of this subtask
		final List<KafkaTopicPartition> allPartitions = partitionDiscoverer.discoverPartitions();

public List<KafkaTopicPartition> discoverPartitions() throws WakeupException, ClosedException {
		if (!closed && !wakeup) {
			try {
			...
				// (2) eliminate partition that are old partitions or should not be subscribed by this subtask
				if (newDiscoveredPartitions == null || newDiscoveredPartitions.isEmpty()) {
					throw new RuntimeException("Unable to retrieve any partitions with KafkaTopicsDescriptor: " + topicsDescriptor);
				} else {
					Iterator<KafkaTopicPartition> iter = newDiscoveredPartitions.iterator();
					KafkaTopicPartition nextPartition;
					while (iter.hasNext()) {
						nextPartition = iter.next();
						//从之前已经发现的KafkaTopicPartition中移除，其二可以保证仅仅是这个subtask的partition
						if (!setAndCheckDiscoveredPartition(nextPartition)) {
							iter.remove();
						}
					}
				}
				return newDiscoveredPartitions;
			...
	}

public boolean setAndCheckDiscoveredPartition(KafkaTopicPartition partition) {
		if (isUndiscoveredPartition(partition)) {
			discoveredPartitions.add(partition);
			
			//kafkaPartition与indexOfThisSubTask --对应
			return KafkaTopicPartitionAssigner.assign(partition, numParallelSubtasks) == indexOfThisSubtask;
		}
		return false;
	}

public static int assign(KafkaTopicPartition partition, int numParallelSubtasks) {
		int startIndex = ((partition.getTopic().hashCode() * 31) & 0x7FFFFFFF) % numParallelSubtasks;
		// here, the assumption is that the id of Kafka partitions are always ascending
		// starting from 0, and therefore can be used directly as the offset clockwise from the start index
		return (startIndex + partition.getPartition()) % numParallelSubtasks;
	}

public class WordSource extends RichParallelSourceFunction<Tuple2<Long, Long>> {
	
	private Boolean isRun = true;
	
	@Override
	public void run(SourceContext<Tuple2<Long, Long>> ctx) throws Exception {
		int start = 0;
		int numberOfParallelSubtasks = getRuntimeContext().getNumberOfParallelSubtasks();
		while (isRun) {
			start += 1;
			if (start % numberOfParallelSubtasks == getRuntimeContext().getIndexOfThisSubtask()) {
				ctx.collect(new Tuple2<>(
						Long.parseLong(start+""),
						1L));
				Thread.sleep(1000);
				System.out.println("Thread.currentThread().getName()=========== " + Thread.currentThread().getName());
			}
		}
	}
	
	@Override
	public void cancel() {
		isRun = false;
	}
}