flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011ITCase.java
/**
	 * Each instance of FlinkKafkaProducer011 uses it's own pool of transactional ids. After the restore from checkpoint
	 * transactional ids are redistributed across the subtasks. In case of scale down, the surplus transactional ids
	 * are dropped. In case of scale up, new one are generated (for the new subtasks). This test make sure that sequence
	 * of scaling down and up again works fine. Especially it checks whether the newly generated ids in scaling up
	 * do not overlap with ids that were used before scaling down. For example we start with 4 ids and parallelism 4:
	 * [1], [2], [3], [4] - one assigned per each subtask
	 * we scale down to parallelism 2:
	 * [1, 2], [3, 4] - first subtask got id 1 and 2, second got ids 3 and 4
	 * surplus ids are dropped from the pools and we scale up to parallelism 3:
	 * [1 or 2], [3 or 4], [???]
	 * new subtask have to generate new id(s), but he can not use ids that are potentially in use, so it has to generate
	 * new ones that are greater then 4.
	 */
Each instance of FlinkKafkaProducer011 uses it's own pool of transactional ids. After the restore from checkpoint
transactional ids are redistributed across the subtasks. In case of scale down, the surplus transactional ids
are dropped. In case of scale up, new one are generated (for the new subtasks). This test make sure that sequence
of scaling down and up again works fine. Especially it checks whether the newly generated ids in scaling up
do not overlap with ids that were used before scaling down.
Each instance of FlinkKafkaProducer011 uses it ' s own pool of transactional ids . After the restore from checkpoint transactional ids are redistributed across the subtasks . In case of scale down the surplus transactional ids are dropped . In case of scale up new one are generated ( for the new subtasks ) . This test make sure that sequence of scaling down and up again works fine . Especially it checks whether the newly generated ids in scaling up do not overlap with ids that were used before scaling down .
@Test(timeout = 120_000L)
	public void testScaleUpAfterScalingDown() throws Exception {
		String topic = "scale-down-before-first-checkpoint";
		final int parallelism1 = 4;
		final int parallelism2 = 2;
		final int parallelism3 = 3;
		final int maxParallelism = Math.max(parallelism1, Math.max(parallelism2, parallelism3));
		List<OperatorStateHandle> operatorStateHandles = repartitionAndExecute(
			topic,
			Collections.emptyList(),
			parallelism1,
			maxParallelism,
			IntStream.range(0, parallelism1).boxed().iterator());
		operatorStateHandles = repartitionAndExecute(
			topic,
			operatorStateHandles,
			parallelism2,
			maxParallelism,
			IntStream.range(parallelism1,  parallelism1 + parallelism2).boxed().iterator());
		operatorStateHandles = repartitionAndExecute(
			topic,
			operatorStateHandles,
			parallelism3,
			maxParallelism,
			IntStream.range(parallelism1 + parallelism2,  parallelism1 + parallelism2 + parallelism3).boxed().iterator());
		operatorStateHandles = repartitionAndExecute(
			topic,
			operatorStateHandles,
			1,
			maxParallelism,
			Collections.emptyIterator());
		assertExactlyOnceForTopic(
			createProperties(),
			topic,
			0,
			IntStream.range(0, parallelism1 + parallelism2 + parallelism3).boxed().collect(Collectors.toList()),
			30_000L);
		deleteTestTopic(topic);
	}
@ Test ( timeout = 120_000L ) public void testScaleUpAfterScalingDown ( ) throws Exception { String topic = "scale-down-before-first-checkpoint" ; final int parallelism1 = 4 ; final int parallelism2 = 2 ; final int parallelism3 = 3 ; final int maxParallelism = Math . max ( parallelism1 , Math . max ( parallelism2 , parallelism3 ) ) ; List < OperatorStateHandle > operatorStateHandles = repartitionAndExecute ( topic , Collections . emptyList ( ) , parallelism1 , maxParallelism , IntStream . range ( 0 , parallelism1 ) . boxed ( ) . iterator ( ) ) ; operatorStateHandles = repartitionAndExecute ( topic , operatorStateHandles , parallelism2 , maxParallelism , IntStream . range ( parallelism1 , parallelism1 + parallelism2 ) . boxed ( ) . iterator ( ) ) ; operatorStateHandles = repartitionAndExecute ( topic , operatorStateHandles , parallelism3 , maxParallelism , IntStream . range ( parallelism1 + parallelism2 , parallelism1 + parallelism2 + parallelism3 ) . boxed ( ) . iterator ( ) ) ; operatorStateHandles = repartitionAndExecute ( topic , operatorStateHandles , 1 , maxParallelism , Collections . emptyIterator ( ) ) ; assertExactlyOnceForTopic ( createProperties ( ) , topic , 0 , IntStream . range ( 0 , parallelism1 + parallelism2 + parallelism3 ) . boxed ( ) . collect ( Collectors . toList ( ) ) , 30_000L ) ; deleteTestTopic ( topic ) ; }
Each instance of FlinkKafkaProducer011 uses it's own pool of transactional ids.
Each instance of FlinkKafkaProducer011 uses it ' s own pool of transactional ids .
// After each previous repartitionAndExecute call, we are left with some lingering transactions, that would // not allow us to read all committed messages from the topic. Thus we initialize operators from // operatorStateHandles once more, but without any new data. This should terminate all ongoing transactions.