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 fewer Flink sources than Kafka partitions, so
	 * one Flink source will read multiple Kafka partitions.
	 */
Tests the proper consumption when having fewer Flink sources than Kafka partitions, so
one Flink source will read multiple Kafka partitions.
Tests the proper consumption when having fewer Flink sources than Kafka partitions so one Flink source will read multiple Kafka partitions .
public void runOneSourceMultiplePartitionsExactlyOnceTest() throws Exception {
		final String topic = "oneToManyTopic";
		final int numPartitions = 5;
		final int numElementsPerPartition = 1000;
		final int totalElements = numPartitions * numElementsPerPartition;
		final int failAfterElements = numElementsPerPartition / 3;
		final int parallelism = 2;
		createTestTopic(topic, numPartitions, 1);
		DataGenerators.generateRandomizedIntegerSequence(
				StreamExecutionEnvironment.getExecutionEnvironment(),
				kafkaServer,
				topic,
				numPartitions,
				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(numPartitions, 3))
				.map(new FailingIdentityMapper<Integer>(failAfterElements))
				.addSink(new ValidatingExactlyOnceSink(totalElements)).setParallelism(1);
		FailingIdentityMapper.failedBefore = false;
		tryExecute(env, "One-source-multi-partitions exactly once test");
		deleteTestTopic(topic);
	}
public void runOneSourceMultiplePartitionsExactlyOnceTest ( ) throws Exception { final String topic = "oneToManyTopic" ; final int numPartitions = 5 ; final int numElementsPerPartition = 1000 ; final int totalElements = numPartitions * numElementsPerPartition ; final int failAfterElements = numElementsPerPartition / 3 ; final int parallelism = 2 ; createTestTopic ( topic , numPartitions , 1 ) ; DataGenerators . generateRandomizedIntegerSequence ( StreamExecutionEnvironment . getExecutionEnvironment ( ) , kafkaServer , topic , numPartitions , 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 ( numPartitions , 3 ) ) . map ( new FailingIdentityMapper < Integer > ( failAfterElements ) ) . addSink ( new ValidatingExactlyOnceSink ( totalElements ) ) . setParallelism ( 1 ) ; FailingIdentityMapper . failedBefore = false ; tryExecute ( env , "One-source-multi-partitions exactly once test" ) ; deleteTestTopic ( topic ) ; }
Tests the proper consumption when having fewer Flink sources than Kafka partitions, so
 one Flink source will read multiple Kafka partitions.
Tests the proper consumption when having fewer Flink sources than Kafka partitions so one Flink source will read multiple Kafka partitions .
// run the topology that fails and recovers