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
/**
	 * This test ensures that transactions reusing transactional.ids (after returning to the pool) will not clash
	 * with previous transactions using same transactional.ids.
	 */
This test ensures that transactions reusing transactional.ids (after returning to the pool) will not clash
with previous transactions using same transactional.ids.
This test ensures that transactions reusing transactional . ids ( after returning to the pool ) will not clash with previous transactions using same transactional . ids .
@Test(timeout = 120_000L)
	public void testRestoreToCheckpointAfterExceedingProducersPool() throws Exception {
		String topic = "flink-kafka-producer-fail-before-notify";
		try (OneInputStreamOperatorTestHarness<Integer, Object> testHarness1 = createTestHarness(topic)) {
			testHarness1.setup();
			testHarness1.open();
			testHarness1.processElement(42, 0);
			OperatorStateHandles snapshot = testHarness1.snapshot(0, 0);
			testHarness1.processElement(43, 0);
			testHarness1.notifyOfCompletedCheckpoint(0);
			try {
				for (int i = 0; i < FlinkKafkaProducer011.DEFAULT_KAFKA_PRODUCERS_POOL_SIZE; i++) {
					testHarness1.snapshot(i + 1, 0);
					testHarness1.processElement(i, 0);
				}
				throw new IllegalStateException("This should not be reached.");
			}
			catch (Exception ex) {
				assertIsCausedBy(FlinkKafka011ErrorCode.PRODUCERS_POOL_EMPTY, ex);
			}
			try (OneInputStreamOperatorTestHarness<Integer, Object> testHarness2 = createTestHarness(topic)) {
				testHarness2.setup();
				testHarness2.initializeState(snapshot);
				testHarness2.open();
			}
			assertExactlyOnceForTopic(createProperties(), topic, 0, Arrays.asList(42), 30_000L);
			deleteTestTopic(topic);
		}
		catch (Exception ex) {
			assertIsCausedBy(ProducerFencedException.class, ex);
		}
	}
@ Test ( timeout = 120_000L ) public void testRestoreToCheckpointAfterExceedingProducersPool ( ) throws Exception { String topic = "flink-kafka-producer-fail-before-notify" ; try ( OneInputStreamOperatorTestHarness < Integer , Object > testHarness1 = createTestHarness ( topic ) ) { testHarness1 . setup ( ) ; testHarness1 . open ( ) ; testHarness1 . processElement ( 42 , 0 ) ; OperatorStateHandles snapshot = testHarness1 . snapshot ( 0 , 0 ) ; testHarness1 . processElement ( 43 , 0 ) ; testHarness1 . notifyOfCompletedCheckpoint ( 0 ) ; try { for ( int i = 0 ; i < FlinkKafkaProducer011 . DEFAULT_KAFKA_PRODUCERS_POOL_SIZE ; i ++ ) { testHarness1 . snapshot ( i + 1 , 0 ) ; testHarness1 . processElement ( i , 0 ) ; } throw new IllegalStateException ( "This should not be reached." ) ; } catch ( Exception ex ) { assertIsCausedBy ( FlinkKafka011ErrorCode . PRODUCERS_POOL_EMPTY , ex ) ; } try ( OneInputStreamOperatorTestHarness < Integer , Object > testHarness2 = createTestHarness ( topic ) ) { testHarness2 . setup ( ) ; testHarness2 . initializeState ( snapshot ) ; testHarness2 . open ( ) ; } assertExactlyOnceForTopic ( createProperties ( ) , topic , 0 , Arrays . asList ( 42 ) , 30_000L ) ; deleteTestTopic ( topic ) ; } catch ( Exception ex ) { assertIsCausedBy ( ProducerFencedException . class , ex ) ; } }
This test ensures that transactions reusing transactional.ids (after returning to the pool) will not clash
 with previous transactions using same transactional.ids.
This test ensures that transactions reusing transactional . ids ( after returning to the pool ) will not clash with previous transactions using same transactional . ids .
// Resume transactions before testHrness1 is being closed (in case of failures close() might not be called) // restore from snapshot1, transactions with records 43 and 44 should be aborted // testHarness1 will be fenced off after creating and closing testHarness2