flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011Tests.java
/**
	 * This tests checks whether FlinkKafkaProducer011 correctly aborts lingering transactions after a failure,
	 * which happened before first checkpoint and was followed up by reducing the parallelism.
	 * 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,
which happened before first checkpoint and was followed up by reducing the parallelism.
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 which happened before first checkpoint and was followed up by reducing the parallelism . 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 testScaleDownBeforeFirstCheckpoint() throws Exception {
		String topic = "scale-down-before-first-checkpoint";
		List<AutoCloseable> operatorsToClose = new ArrayList<>();
		int preScaleDownParallelism = Math.max(2, FlinkKafkaProducer011.SAFE_SCALE_DOWN_FACTOR);
		for (int subtaskIndex = 0; subtaskIndex < preScaleDownParallelism; subtaskIndex++) {
			OneInputStreamOperatorTestHarness<Integer, Object> preScaleDownOperator = createTestHarness(
				topic,
				preScaleDownParallelism,
				preScaleDownParallelism,
				subtaskIndex);
			preScaleDownOperator.setup();
			preScaleDownOperator.open();
			preScaleDownOperator.processElement(subtaskIndex * 2, 0);
			preScaleDownOperator.snapshot(0, 1);
			preScaleDownOperator.processElement(subtaskIndex * 2 + 1, 2);
			operatorsToClose.add(preScaleDownOperator);
		}
		OneInputStreamOperatorTestHarness<Integer, Object> postScaleDownOperator1 = createTestHarness(topic, 1, 1, 0);
		postScaleDownOperator1.setup();
		postScaleDownOperator1.open();
		postScaleDownOperator1.processElement(46, 7);
		postScaleDownOperator1.snapshot(4, 8);
		postScaleDownOperator1.processElement(47, 9);
		postScaleDownOperator1.notifyOfCompletedCheckpoint(4);
		assertExactlyOnceForTopic(createProperties(), topic, 0, Arrays.asList(46), 30_000L);
		postScaleDownOperator1.close();
		for (AutoCloseable operatorToClose : operatorsToClose) {
			closeIgnoringProducerFenced(operatorToClose);
		}
		deleteTestTopic(topic);
	}
@ Test ( timeout = 120_000L ) public void testScaleDownBeforeFirstCheckpoint ( ) throws Exception { String topic = "scale-down-before-first-checkpoint" ; List < AutoCloseable > operatorsToClose = new ArrayList < > ( ) ; int preScaleDownParallelism = Math . max ( 2 , FlinkKafkaProducer011 . SAFE_SCALE_DOWN_FACTOR ) ; for ( int subtaskIndex = 0 ; subtaskIndex < preScaleDownParallelism ; subtaskIndex ++ ) { OneInputStreamOperatorTestHarness < Integer , Object > preScaleDownOperator = createTestHarness ( topic , preScaleDownParallelism , preScaleDownParallelism , subtaskIndex ) ; preScaleDownOperator . setup ( ) ; preScaleDownOperator . open ( ) ; preScaleDownOperator . processElement ( subtaskIndex * 2 , 0 ) ; preScaleDownOperator . snapshot ( 0 , 1 ) ; preScaleDownOperator . processElement ( subtaskIndex * 2 + 1 , 2 ) ; operatorsToClose . add ( preScaleDownOperator ) ; } OneInputStreamOperatorTestHarness < Integer , Object > postScaleDownOperator1 = createTestHarness ( topic , 1 , 1 , 0 ) ; postScaleDownOperator1 . setup ( ) ; postScaleDownOperator1 . open ( ) ; postScaleDownOperator1 . processElement ( 46 , 7 ) ; postScaleDownOperator1 . snapshot ( 4 , 8 ) ; postScaleDownOperator1 . processElement ( 47 , 9 ) ; postScaleDownOperator1 . notifyOfCompletedCheckpoint ( 4 ) ; assertExactlyOnceForTopic ( createProperties ( ) , topic , 0 , Arrays . asList ( 46 ) , 30_000L ) ; postScaleDownOperator1 . close ( ) ; for ( AutoCloseable operatorToClose : operatorsToClose ) { closeIgnoringProducerFenced ( operatorToClose ) ; } deleteTestTopic ( topic ) ; }
This tests checks whether FlinkKafkaProducer011 correctly aborts lingering transactions after a failure,
 which happened before first checkpoint and was followed up by reducing the parallelism.
This tests checks whether FlinkKafkaProducer011 correctly aborts lingering transactions after a failure which happened before first checkpoint and was followed up by reducing the parallelism .
// do not close previous testHarnesses to make sure that closing do not clean up something (in case of failure // there might not be any close) // After previous failure simulate restarting application with smaller parallelism // write and commit more records, after potentially lingering transactions //now we should have: // - records 42, 43, 44 and 45 in aborted transactions // - committed transaction with record 46 // - pending transaction with record 47 // ignore ProducerFencedExceptions, because postScaleDownOperator1 could reuse transactional ids.