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/KafkaConsumerTestBase.java
/**
	 * Tests the proper consumption when having a 1:1 correspondence between kafka partitions and
	 * Flink sources.
	 */
Tests the proper consumption when having a 1:1 correspondence between kafka partitions and
Flink sources.
Tests the proper consumption when having a 1 : 1 correspondence between kafka partitions and Flink sources .
public void runOneToOneExactlyOnceTest() throws Exception {
		final String topic = "oneToOneTopic";
		final int parallelism = 5;
		final int numElementsPerPartition = 1000;
		final int totalElements = parallelism * numElementsPerPartition;
		final int failAfterElements = numElementsPerPartition / 3;
		createTestTopic(topic, parallelism, 1);
		DataGenerators.generateRandomizedIntegerSequence(
				StreamExecutionEnvironment.getExecutionEnvironment(),
				kafkaServer,
				topic,
				parallelism,
				numElementsPerPartition,
				true);
		DeserializationSchema<Integer> schema =
				new TypeInformationSerializationSchema<>(BasicTypeInfo.INT_TYPE_INFO, new ExecutionConfig());
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.enableCheckpointing(500);
		env.setParallelism(parallelism);
		env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, 0));
		env.getConfig().disableSysoutLogging();
		Properties props = new Properties();
		props.putAll(standardProps);
		props.putAll(secureProps);
		FlinkKafkaConsumerBase<Integer> kafkaSource = kafkaServer.getConsumer(topic, schema, props);
		env
				.addSource(kafkaSource)
				.map(new PartitionValidatingMapper(parallelism, 1))
				.map(new FailingIdentityMapper<Integer>(failAfterElements))
				.addSink(new ValidatingExactlyOnceSink(totalElements)).setParallelism(1);
		FailingIdentityMapper.failedBefore = false;
		tryExecute(env, "One-to-one exactly once test");
		deleteTestTopic(topic);
	}
public void runOneToOneExactlyOnceTest ( ) throws Exception { final String topic = "oneToOneTopic" ; final int parallelism = 5 ; final int numElementsPerPartition = 1000 ; final int totalElements = parallelism * numElementsPerPartition ; final int failAfterElements = numElementsPerPartition / 3 ; createTestTopic ( topic , parallelism , 1 ) ; DataGenerators . generateRandomizedIntegerSequence ( StreamExecutionEnvironment . getExecutionEnvironment ( ) , kafkaServer , topic , parallelism , numElementsPerPartition , true ) ; DeserializationSchema < Integer > schema = new TypeInformationSerializationSchema < > ( BasicTypeInfo . INT_TYPE_INFO , new ExecutionConfig ( ) ) ; StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env . enableCheckpointing ( 500 ) ; env . setParallelism ( parallelism ) ; env . setRestartStrategy ( RestartStrategies . fixedDelayRestart ( 1 , 0 ) ) ; env . getConfig ( ) . disableSysoutLogging ( ) ; Properties props = new Properties ( ) ; props . putAll ( standardProps ) ; props . putAll ( secureProps ) ; FlinkKafkaConsumerBase < Integer > kafkaSource = kafkaServer . getConsumer ( topic , schema , props ) ; env . addSource ( kafkaSource ) . map ( new PartitionValidatingMapper ( parallelism , 1 ) ) . map ( new FailingIdentityMapper < Integer > ( failAfterElements ) ) . addSink ( new ValidatingExactlyOnceSink ( totalElements ) ) . setParallelism ( 1 ) ; FailingIdentityMapper . failedBefore = false ; tryExecute ( env , "One-to-one exactly once test" ) ; deleteTestTopic ( topic ) ; }
Tests the proper consumption when having a 1:1 correspondence between kafka partitions and
 Flink sources.
Tests the proper consumption when having a 1 : 1 correspondence between kafka partitions and Flink sources .
// run the topology that fails and recovers