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 more Flink sources than Kafka partitions, which means
	 * that some Flink sources will read no partitions.
	 */
Tests the proper consumption when having more Flink sources than Kafka partitions, which means
that some Flink sources will read no partitions.
Tests the proper consumption when having more Flink sources than Kafka partitions which means that some Flink sources will read no partitions .
public void runMultipleSourcesOnePartitionExactlyOnceTest() throws Exception {
		final String topic = "manyToOneTopic";
		final int numPartitions = 5;
		final int numElementsPerPartition = 1000;
		final int totalElements = numPartitions * numElementsPerPartition;
		final int failAfterElements = numElementsPerPartition / 3;
		final int parallelism = 8;
		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();
		env.setBufferTimeout(0);
		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, 1))
			.map(new FailingIdentityMapper<Integer>(failAfterElements))
			.addSink(new ValidatingExactlyOnceSink(totalElements)).setParallelism(1);
		FailingIdentityMapper.failedBefore = false;
		tryExecute(env, "multi-source-one-partitions exactly once test");
		deleteTestTopic(topic);
	}
public void runMultipleSourcesOnePartitionExactlyOnceTest ( ) throws Exception { final String topic = "manyToOneTopic" ; final int numPartitions = 5 ; final int numElementsPerPartition = 1000 ; final int totalElements = numPartitions * numElementsPerPartition ; final int failAfterElements = numElementsPerPartition / 3 ; final int parallelism = 8 ; 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 ( ) ; env . setBufferTimeout ( 0 ) ; 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 , 1 ) ) . map ( new FailingIdentityMapper < Integer > ( failAfterElements ) ) . addSink ( new ValidatingExactlyOnceSink ( totalElements ) ) . setParallelism ( 1 ) ; FailingIdentityMapper . failedBefore = false ; tryExecute ( env , "multi-source-one-partitions exactly once test" ) ; deleteTestTopic ( topic ) ; }
Tests the proper consumption when having more Flink sources than Kafka partitions, which means
 that some Flink sources will read no partitions.
Tests the proper consumption when having more Flink sources than Kafka partitions which means that some Flink sources will read no partitions .
// run the topology that fails and recovers // set the number of restarts to one. The failing mapper will fail once, then it's only success exceptions.