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
/**
	 * Ensure Kafka is working on both producer and consumer side.
	 * This executes a job that contains two Flink pipelines.
	 *
	 * <pre>
	 * (generator source) --> (kafka sink)-[KAFKA-TOPIC]-(kafka source) --> (validating sink)
	 * </pre>
	 *
	 * <p>We need to externally retry this test. We cannot let Flink's retry mechanism do it, because the Kafka producer
	 * does not guarantee exactly-once output. Hence a recovery would introduce duplicates that
	 * cause the test to fail.
	 *
	 * <p>This test also ensures that FLINK-3156 doesn't happen again:
	 *
	 * <p>The following situation caused a NPE in the FlinkKafkaConsumer
	 *
	 * <p>topic-1 <-- elements are only produced into topic1.
	 * topic-2
	 *
	 * <p>Therefore, this test is consuming as well from an empty topic.
	 *
	 */
Ensure Kafka is working on both producer and consumer side.
This executes a job that contains two Flink pipelines.



We need to externally retry this test. We cannot let Flink's retry mechanism do it, because the Kafka producer
does not guarantee exactly-once output. Hence a recovery would introduce duplicates that
cause the test to fail.

This test also ensures that FLINK-3156 doesn't happen again.

The following situation caused a NPE in the FlinkKafkaConsumer

topic-1 <-- elements are only produced into topic1.
topic-2

Therefore, this test is consuming as well from an empty topic.
Ensure Kafka is working on both producer and consumer side . This executes a job that contains two Flink pipelines . We need to externally retry this test . We cannot let Flink ' s retry mechanism do it because the Kafka producer does not guarantee exactly - once output . Hence a recovery would introduce duplicates that cause the test to fail . This test also ensures that FLINK - 3156 doesn ' t happen again . The following situation caused a NPE in the FlinkKafkaConsumer topic - 1 < -- elements are only produced into topic1 . topic - 2 Therefore this test is consuming as well from an empty topic .
@RetryOnException(times = 2, exception = kafka.common.NotLeaderForPartitionException.class)
	public void runSimpleConcurrentProducerConsumerTopology() throws Exception {
		final String topic = "concurrentProducerConsumerTopic_" + UUID.randomUUID().toString();
		final String additionalEmptyTopic = "additionalEmptyTopic_" + UUID.randomUUID().toString();
		final int parallelism = 3;
		final int elementsPerPartition = 100;
		final int totalElements = parallelism * elementsPerPartition;
		createTestTopic(topic, parallelism, 2);
		createTestTopic(additionalEmptyTopic, parallelism, 1); 
		final StreamExecutionEnvironment env =
				StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(parallelism);
		env.enableCheckpointing(500);
		env.setRestartStrategy(RestartStrategies.noRestart()); 
		env.getConfig().disableSysoutLogging();
		TypeInformation<Tuple2<Long, String>> longStringType = TypeInfoParser.parse("Tuple2<Long, String>");
		TypeInformationSerializationSchema<Tuple2<Long, String>> sourceSchema =
				new TypeInformationSerializationSchema<>(longStringType, env.getConfig());
		TypeInformationSerializationSchema<Tuple2<Long, String>> sinkSchema =
				new TypeInformationSerializationSchema<>(longStringType, env.getConfig());
		DataStream<Tuple2<Long, String>> stream = env.addSource(new RichParallelSourceFunction<Tuple2<Long, String>>() {
			private boolean running = true;
			@Override
			public void run(SourceContext<Tuple2<Long, String>> ctx) throws InterruptedException {
				int cnt = getRuntimeContext().getIndexOfThisSubtask() * elementsPerPartition;
				int limit = cnt + elementsPerPartition;
				while (running && cnt < limit) {
					ctx.collect(new Tuple2<>(1000L + cnt, "kafka-" + cnt));
					cnt++;
					Thread.sleep(50);
				}
			}
			@Override
			public void cancel() {
				running = false;
			}
		});
		Properties producerProperties = FlinkKafkaProducerBase.getPropertiesFromBrokerList(brokerConnectionStrings);
		producerProperties.setProperty("retries", "3");
		producerProperties.putAll(secureProps);
		kafkaServer.produceIntoKafka(stream, topic, new KeyedSerializationSchemaWrapper<>(sinkSchema), producerProperties, null);
		List<String> topics = new ArrayList<>();
		topics.add(topic);
		topics.add(additionalEmptyTopic);
		Properties props = new Properties();
		props.putAll(standardProps);
		props.putAll(secureProps);
		FlinkKafkaConsumerBase<Tuple2<Long, String>> source = kafkaServer.getConsumer(topics, sourceSchema, props);
		DataStreamSource<Tuple2<Long, String>> consuming = env.addSource(source).setParallelism(parallelism);
		consuming.addSink(new RichSinkFunction<Tuple2<Long, String>>() {
			private int elCnt = 0;
			private BitSet validator = new BitSet(totalElements);
			@Override
			public void invoke(Tuple2<Long, String> value) throws Exception {
				String[] sp = value.f1.split("-");
				int v = Integer.parseInt(sp[1]);
				assertEquals(value.f0 - 1000, (long) v);
				assertFalse("Received tuple twice", validator.get(v));
				validator.set(v);
				elCnt++;
				if (elCnt == totalElements) {
					int nc;
					if ((nc = validator.nextClearBit(0)) != totalElements) {
						fail("The bitset was not set to 1 on all elements. Next clear:"
								+ nc + " Set: " + validator);
					}
					throw new SuccessException();
				}
			}
			@Override
			public void close() throws Exception {
				super.close();
			}
		}).setParallelism(1);
		try {
			tryExecutePropagateExceptions(env, "runSimpleConcurrentProducerConsumerTopology");
		}
		catch (ProgramInvocationException | JobExecutionException e) {
			Throwable cause = e.getCause();
			int depth = 0;
			while (cause != null && depth++ < 20) {
				if (cause instanceof kafka.common.NotLeaderForPartitionException) {
					throw (Exception) cause;
				}
				cause = cause.getCause();
			}
			throw e;
		}
		deleteTestTopic(topic);
	}
@ RetryOnException ( times = 2 , exception = kafka . common . NotLeaderForPartitionException . class ) public void runSimpleConcurrentProducerConsumerTopology ( ) throws Exception { final String topic = "concurrentProducerConsumerTopic_" + UUID . randomUUID ( ) . toString ( ) ; final String additionalEmptyTopic = "additionalEmptyTopic_" + UUID . randomUUID ( ) . toString ( ) ; final int parallelism = 3 ; final int elementsPerPartition = 100 ; final int totalElements = parallelism * elementsPerPartition ; createTestTopic ( topic , parallelism , 2 ) ; createTestTopic ( additionalEmptyTopic , parallelism , 1 ) ; final StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env . setParallelism ( parallelism ) ; env . enableCheckpointing ( 500 ) ; env . setRestartStrategy ( RestartStrategies . noRestart ( ) ) ; env . getConfig ( ) . disableSysoutLogging ( ) ; TypeInformation < Tuple2 < Long , String > > longStringType = TypeInfoParser . parse ( "Tuple2<Long, String>" ) ; TypeInformationSerializationSchema < Tuple2 < Long , String > > sourceSchema = new TypeInformationSerializationSchema < > ( longStringType , env . getConfig ( ) ) ; TypeInformationSerializationSchema < Tuple2 < Long , String > > sinkSchema = new TypeInformationSerializationSchema < > ( longStringType , env . getConfig ( ) ) ; DataStream < Tuple2 < Long , String > > stream = env . addSource ( new RichParallelSourceFunction < Tuple2 < Long , String > > ( ) { private boolean running = true ; @ Override public void run ( SourceContext < Tuple2 < Long , String > > ctx ) throws InterruptedException { int cnt = getRuntimeContext ( ) . getIndexOfThisSubtask ( ) * elementsPerPartition ; int limit = cnt + elementsPerPartition ; while ( running && cnt < limit ) { ctx . collect ( new Tuple2 < > ( 1000L + cnt , "kafka-" + cnt ) ) ; cnt ++ ; Thread . sleep ( 50 ) ; } } @ Override public void cancel ( ) { running = false ; } } ) ; Properties producerProperties = FlinkKafkaProducerBase . getPropertiesFromBrokerList ( brokerConnectionStrings ) ; producerProperties . setProperty ( "retries" , "3" ) ; producerProperties . putAll ( secureProps ) ; kafkaServer . produceIntoKafka ( stream , topic , new KeyedSerializationSchemaWrapper < > ( sinkSchema ) , producerProperties , null ) ; List < String > topics = new ArrayList < > ( ) ; topics . add ( topic ) ; topics . add ( additionalEmptyTopic ) ; Properties props = new Properties ( ) ; props . putAll ( standardProps ) ; props . putAll ( secureProps ) ; FlinkKafkaConsumerBase < Tuple2 < Long , String > > source = kafkaServer . getConsumer ( topics , sourceSchema , props ) ; DataStreamSource < Tuple2 < Long , String > > consuming = env . addSource ( source ) . setParallelism ( parallelism ) ; consuming . addSink ( new RichSinkFunction < Tuple2 < Long , String > > ( ) { private int elCnt = 0 ; private BitSet validator = new BitSet ( totalElements ) ; @ Override public void invoke ( Tuple2 < Long , String > value ) throws Exception { String [ ] sp = value . f1 . split ( "-" ) ; int v = Integer . parseInt ( sp [ 1 ] ) ; assertEquals ( value . f0 - 1000 , ( long ) v ) ; assertFalse ( "Received tuple twice" , validator . get ( v ) ) ; validator . set ( v ) ; elCnt ++ ; if ( elCnt == totalElements ) { int nc ; if ( ( nc = validator . nextClearBit ( 0 ) ) != totalElements ) { fail ( "The bitset was not set to 1 on all elements. Next clear:" + nc + " Set: " + validator ) ; } throw new SuccessException ( ) ; } } @ Override public void close ( ) throws Exception { super . close ( ) ; } } ) . setParallelism ( 1 ) ; try { tryExecutePropagateExceptions ( env , "runSimpleConcurrentProducerConsumerTopology" ) ; } catch ( ProgramInvocationException | JobExecutionException e ) { Throwable cause = e . getCause ( ) ; int depth = 0 ; while ( cause != null && depth ++ < 20 ) { if ( cause instanceof kafka . common . NotLeaderForPartitionException ) { throw ( Exception ) cause ; } cause = cause . getCause ( ) ; } throw e ; } deleteTestTopic ( topic ) ; }
Ensure Kafka is working on both producer and consumer side.
Ensure Kafka is working on both producer and consumer side .
// create an empty topic which will remain empty all the time // fail immediately // ----------- add producer dataflow ---------- // we delay data generation a bit so that we are sure that some checkpoints are // triggered (for FLINK-3156) // ----------- add consumer dataflow ---------- // check if everything in the bitset is set to true // look for NotLeaderForPartitionException // search for nested SuccessExceptions