flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
/**
	 * Tests whether the Kafka consumer behaves correctly when scaling the parallelism up/down,
	 * which means that operator state is being reshuffled.
	 *
	 * <p>This also verifies that a restoring source is always impervious to changes in the list
	 * of topics fetched from Kafka.
	 */
Tests whether the Kafka consumer behaves correctly when scaling the parallelism up/down,
which means that operator state is being reshuffled.

This also verifies that a restoring source is always impervious to changes in the list
of topics fetched from Kafka.
Tests whether the Kafka consumer behaves correctly when scaling the parallelism up / down which means that operator state is being reshuffled . This also verifies that a restoring source is always impervious to changes in the list of topics fetched from Kafka .
@SuppressWarnings("unchecked")
	private void testRescaling(
		final int initialParallelism,
		final int numPartitions,
		final int restoredParallelism,
		final int restoredNumPartitions) throws Exception {
		Preconditions.checkArgument(
			restoredNumPartitions >= numPartitions,
			"invalid test case for Kafka repartitioning; Kafka only allows increasing partitions.");
		List<KafkaTopicPartition> mockFetchedPartitionsOnStartup = new ArrayList<>();
		for (int i = 0; i < numPartitions; i++) {
			mockFetchedPartitionsOnStartup.add(new KafkaTopicPartition("test-topic", i));
		}
		DummyFlinkKafkaConsumer<String>[] consumers =
			new DummyFlinkKafkaConsumer[initialParallelism];
		AbstractStreamOperatorTestHarness<String>[] testHarnesses =
			new AbstractStreamOperatorTestHarness[initialParallelism];
		for (int i = 0; i < initialParallelism; i++) {
			TestPartitionDiscoverer partitionDiscoverer = new TestPartitionDiscoverer(
				new KafkaTopicsDescriptor(Collections.singletonList("test-topic"), null),
				i,
				initialParallelism,
				TestPartitionDiscoverer.createMockGetAllTopicsSequenceFromFixedReturn(Collections.singletonList("test-topic")),
				TestPartitionDiscoverer.createMockGetAllPartitionsFromTopicsSequenceFromFixedReturn(mockFetchedPartitionsOnStartup));
			consumers[i] = new DummyFlinkKafkaConsumer<>(mock(AbstractFetcher.class), partitionDiscoverer, false);
			testHarnesses[i] = createTestHarness(consumers[i], initialParallelism, i);
			testHarnesses[i].initializeState(null);
			testHarnesses[i].open();
		}
		Map<KafkaTopicPartition, Long> globalSubscribedPartitions = new HashMap<>();
		for (int i = 0; i < initialParallelism; i++) {
			Map<KafkaTopicPartition, Long> subscribedPartitions =
				consumers[i].getSubscribedPartitionsToStartOffsets();
			for (KafkaTopicPartition partition : subscribedPartitions.keySet()) {
				assertThat(globalSubscribedPartitions, not(hasKey(partition)));
			}
			globalSubscribedPartitions.putAll(subscribedPartitions);
		}
		assertThat(globalSubscribedPartitions.values(), hasSize(numPartitions));
		assertThat(mockFetchedPartitionsOnStartup, everyItem(isIn(globalSubscribedPartitions.keySet())));
		OperatorStateHandles[] state = new OperatorStateHandles[initialParallelism];
		for (int i = 0; i < initialParallelism; i++) {
			state[i] = testHarnesses[i].snapshot(0, 0);
		}
		OperatorStateHandles mergedState = AbstractStreamOperatorTestHarness.repackageState(state);
		List<KafkaTopicPartition> mockFetchedPartitionsAfterRestore = new ArrayList<>();
		for (int i = 0; i < restoredNumPartitions; i++) {
			mockFetchedPartitionsAfterRestore.add(new KafkaTopicPartition("test-topic", i));
		}
		DummyFlinkKafkaConsumer<String>[] restoredConsumers =
			new DummyFlinkKafkaConsumer[restoredParallelism];
		AbstractStreamOperatorTestHarness<String>[] restoredTestHarnesses =
			new AbstractStreamOperatorTestHarness[restoredParallelism];
		for (int i = 0; i < restoredParallelism; i++) {
			TestPartitionDiscoverer partitionDiscoverer = new TestPartitionDiscoverer(
				new KafkaTopicsDescriptor(Collections.singletonList("test-topic"), null),
				i,
				restoredParallelism,
				TestPartitionDiscoverer.createMockGetAllTopicsSequenceFromFixedReturn(Collections.singletonList("test-topic")),
				TestPartitionDiscoverer.createMockGetAllPartitionsFromTopicsSequenceFromFixedReturn(mockFetchedPartitionsAfterRestore));
			restoredConsumers[i] = new DummyFlinkKafkaConsumer<>(mock(AbstractFetcher.class), partitionDiscoverer, false);
			restoredTestHarnesses[i] = createTestHarness(restoredConsumers[i], restoredParallelism, i);
			restoredTestHarnesses[i].initializeState(mergedState);
			restoredTestHarnesses[i].open();
		}
		Map<KafkaTopicPartition, Long> restoredGlobalSubscribedPartitions = new HashMap<>();
		for (int i = 0; i < restoredParallelism; i++) {
			Map<KafkaTopicPartition, Long> subscribedPartitions =
				restoredConsumers[i].getSubscribedPartitionsToStartOffsets();
			for (KafkaTopicPartition partition : subscribedPartitions.keySet()) {
				assertThat(restoredGlobalSubscribedPartitions, not(hasKey(partition)));
			}
			restoredGlobalSubscribedPartitions.putAll(subscribedPartitions);
		}
		assertThat(restoredGlobalSubscribedPartitions.values(), hasSize(restoredNumPartitions));
		assertThat(mockFetchedPartitionsOnStartup, everyItem(isIn(restoredGlobalSubscribedPartitions.keySet())));
	}
@ SuppressWarnings ( "unchecked" ) private void testRescaling ( final int initialParallelism , final int numPartitions , final int restoredParallelism , final int restoredNumPartitions ) throws Exception { Preconditions . checkArgument ( restoredNumPartitions >= numPartitions , "invalid test case for Kafka repartitioning; Kafka only allows increasing partitions." ) ; List < KafkaTopicPartition > mockFetchedPartitionsOnStartup = new ArrayList < > ( ) ; for ( int i = 0 ; i < numPartitions ; i ++ ) { mockFetchedPartitionsOnStartup . add ( new KafkaTopicPartition ( "test-topic" , i ) ) ; } DummyFlinkKafkaConsumer < String > [ ] consumers = new DummyFlinkKafkaConsumer [ initialParallelism ] ; AbstractStreamOperatorTestHarness < String > [ ] testHarnesses = new AbstractStreamOperatorTestHarness [ initialParallelism ] ; for ( int i = 0 ; i < initialParallelism ; i ++ ) { TestPartitionDiscoverer partitionDiscoverer = new TestPartitionDiscoverer ( new KafkaTopicsDescriptor ( Collections . singletonList ( "test-topic" ) , null ) , i , initialParallelism , TestPartitionDiscoverer . createMockGetAllTopicsSequenceFromFixedReturn ( Collections . singletonList ( "test-topic" ) ) , TestPartitionDiscoverer . createMockGetAllPartitionsFromTopicsSequenceFromFixedReturn ( mockFetchedPartitionsOnStartup ) ) ; consumers [ i ] = new DummyFlinkKafkaConsumer < > ( mock ( AbstractFetcher . class ) , partitionDiscoverer , false ) ; testHarnesses [ i ] = createTestHarness ( consumers [ i ] , initialParallelism , i ) ; testHarnesses [ i ] . initializeState ( null ) ; testHarnesses [ i ] . open ( ) ; } Map < KafkaTopicPartition , Long > globalSubscribedPartitions = new HashMap < > ( ) ; for ( int i = 0 ; i < initialParallelism ; i ++ ) { Map < KafkaTopicPartition , Long > subscribedPartitions = consumers [ i ] . getSubscribedPartitionsToStartOffsets ( ) ; for ( KafkaTopicPartition partition : subscribedPartitions . keySet ( ) ) { assertThat ( globalSubscribedPartitions , not ( hasKey ( partition ) ) ) ; } globalSubscribedPartitions . putAll ( subscribedPartitions ) ; } assertThat ( globalSubscribedPartitions . values ( ) , hasSize ( numPartitions ) ) ; assertThat ( mockFetchedPartitionsOnStartup , everyItem ( isIn ( globalSubscribedPartitions . keySet ( ) ) ) ) ; OperatorStateHandles [ ] state = new OperatorStateHandles [ initialParallelism ] ; for ( int i = 0 ; i < initialParallelism ; i ++ ) { state [ i ] = testHarnesses [ i ] . snapshot ( 0 , 0 ) ; } OperatorStateHandles mergedState = AbstractStreamOperatorTestHarness . repackageState ( state ) ; List < KafkaTopicPartition > mockFetchedPartitionsAfterRestore = new ArrayList < > ( ) ; for ( int i = 0 ; i < restoredNumPartitions ; i ++ ) { mockFetchedPartitionsAfterRestore . add ( new KafkaTopicPartition ( "test-topic" , i ) ) ; } DummyFlinkKafkaConsumer < String > [ ] restoredConsumers = new DummyFlinkKafkaConsumer [ restoredParallelism ] ; AbstractStreamOperatorTestHarness < String > [ ] restoredTestHarnesses = new AbstractStreamOperatorTestHarness [ restoredParallelism ] ; for ( int i = 0 ; i < restoredParallelism ; i ++ ) { TestPartitionDiscoverer partitionDiscoverer = new TestPartitionDiscoverer ( new KafkaTopicsDescriptor ( Collections . singletonList ( "test-topic" ) , null ) , i , restoredParallelism , TestPartitionDiscoverer . createMockGetAllTopicsSequenceFromFixedReturn ( Collections . singletonList ( "test-topic" ) ) , TestPartitionDiscoverer . createMockGetAllPartitionsFromTopicsSequenceFromFixedReturn ( mockFetchedPartitionsAfterRestore ) ) ; restoredConsumers [ i ] = new DummyFlinkKafkaConsumer < > ( mock ( AbstractFetcher . class ) , partitionDiscoverer , false ) ; restoredTestHarnesses [ i ] = createTestHarness ( restoredConsumers [ i ] , restoredParallelism , i ) ; restoredTestHarnesses [ i ] . initializeState ( mergedState ) ; restoredTestHarnesses [ i ] . open ( ) ; } Map < KafkaTopicPartition , Long > restoredGlobalSubscribedPartitions = new HashMap < > ( ) ; for ( int i = 0 ; i < restoredParallelism ; i ++ ) { Map < KafkaTopicPartition , Long > subscribedPartitions = restoredConsumers [ i ] . getSubscribedPartitionsToStartOffsets ( ) ; for ( KafkaTopicPartition partition : subscribedPartitions . keySet ( ) ) { assertThat ( restoredGlobalSubscribedPartitions , not ( hasKey ( partition ) ) ) ; } restoredGlobalSubscribedPartitions . putAll ( subscribedPartitions ) ; } assertThat ( restoredGlobalSubscribedPartitions . values ( ) , hasSize ( restoredNumPartitions ) ) ; assertThat ( mockFetchedPartitionsOnStartup , everyItem ( isIn ( restoredGlobalSubscribedPartitions . keySet ( ) ) ) ) ; }
Tests whether the Kafka consumer behaves correctly when scaling the parallelism up/down,
 which means that operator state is being reshuffled.
Tests whether the Kafka consumer behaves correctly when scaling the parallelism up / down which means that operator state is being reshuffled .
// initializeState() is always called, null signals that we didn't restore // make sure that no one else is subscribed to these partitions // ----------------------------------------------------------------------------------------- // restore // initializeState() is always called, null signals that we didn't restore // make sure that no one else is subscribed to these partitions