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 tests checks whether FlinkKafkaProducer011 correctly aborts lingering transactions after a failure.
	 * If such transactions were left alone lingering it consumers would be unable to read committed records
	 * that were created after this lingering transaction.
	 */
This tests checks whether FlinkKafkaProducer011 correctly aborts lingering transactions after a failure.
If such transactions were left alone lingering it consumers would be unable to read committed records
that were created after this lingering transaction.
This tests checks whether FlinkKafkaProducer011 correctly aborts lingering transactions after a failure . If such transactions were left alone lingering it consumers would be unable to read committed records that were created after this lingering transaction .
@Test(timeout = 120_000L)
	public void testFailBeforeNotifyAndResumeWorkAfterwards() throws Exception {
		String topic = "flink-kafka-producer-fail-before-notify";
		OneInputStreamOperatorTestHarness<Integer, Object> testHarness = createTestHarness(topic);
		testHarness.setup();
		testHarness.open();
		testHarness.processElement(42, 0);
		testHarness.snapshot(0, 1);
		testHarness.processElement(43, 2);
		OperatorStateHandles snapshot1 = testHarness.snapshot(1, 3);
		testHarness.processElement(44, 4);
		testHarness.snapshot(2, 5);
		testHarness.processElement(45, 6);
		testHarness = createTestHarness(topic);
		testHarness.setup();
		testHarness.initializeState(snapshot1);
		testHarness.open();
		testHarness.processElement(46, 7);
		testHarness.snapshot(4, 8);
		testHarness.processElement(47, 9);
		testHarness.notifyOfCompletedCheckpoint(4);
		assertExactlyOnceForTopic(createProperties(), topic, 0, Arrays.asList(42, 43, 46), 30_000L);
		testHarness.close();
		deleteTestTopic(topic);
	}
@ Test ( timeout = 120_000L ) public void testFailBeforeNotifyAndResumeWorkAfterwards ( ) throws Exception { String topic = "flink-kafka-producer-fail-before-notify" ; OneInputStreamOperatorTestHarness < Integer , Object > testHarness = createTestHarness ( topic ) ; testHarness . setup ( ) ; testHarness . open ( ) ; testHarness . processElement ( 42 , 0 ) ; testHarness . snapshot ( 0 , 1 ) ; testHarness . processElement ( 43 , 2 ) ; OperatorStateHandles snapshot1 = testHarness . snapshot ( 1 , 3 ) ; testHarness . processElement ( 44 , 4 ) ; testHarness . snapshot ( 2 , 5 ) ; testHarness . processElement ( 45 , 6 ) ; testHarness = createTestHarness ( topic ) ; testHarness . setup ( ) ; testHarness . initializeState ( snapshot1 ) ; testHarness . open ( ) ; testHarness . processElement ( 46 , 7 ) ; testHarness . snapshot ( 4 , 8 ) ; testHarness . processElement ( 47 , 9 ) ; testHarness . notifyOfCompletedCheckpoint ( 4 ) ; assertExactlyOnceForTopic ( createProperties ( ) , topic , 0 , Arrays . asList ( 42 , 43 , 46 ) , 30_000L ) ; testHarness . close ( ) ; deleteTestTopic ( topic ) ; }
This tests checks whether FlinkKafkaProducer011 correctly aborts lingering transactions after a failure.
This tests checks whether FlinkKafkaProducer011 correctly aborts lingering transactions after a failure .
// do not close previous testHarness to make sure that closing do not clean up something (in case of failure // there might not be any close) // restore from snapshot1, transactions with records 44 and 45 should be aborted // write and commit more records, after potentially lingering transactions //now we should have: // - records 42 and 43 in committed transactions // - aborted transactions with records 44 and 45 // - committed transaction with record 46 // - pending transaction with record 47