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
/**
	 * Test producing and consuming into multiple topics.
	 * @throws Exception
	 */
Test producing and consuming into multiple topics.
@throws Exception
Test producing and consuming into multiple topics . @throws Exception
public void runProduceConsumeMultipleTopics() throws Exception {
		final int numTopics = 5;
		final int numElements = 20;
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.getConfig().disableSysoutLogging();
		final List<String> topics = new ArrayList<>();
		for (int i = 0; i < numTopics; i++) {
			final String topic = "topic-" + i;
			topics.add(topic);
			createTestTopic(topic, i + 1 /*partitions*/, 1);
		}
		env.setParallelism(1);
		DataStream<Tuple3<Integer, Integer, String>> stream = env.addSource(new RichParallelSourceFunction<Tuple3<Integer, Integer, String>>() {
			@Override
			public void run(SourceContext<Tuple3<Integer, Integer, String>> ctx) throws Exception {
				int partition = getRuntimeContext().getIndexOfThisSubtask();
				for (int topicId = 0; topicId < numTopics; topicId++) {
					for (int i = 0; i < numElements; i++) {
						ctx.collect(new Tuple3<>(partition, i, "topic-" + topicId));
					}
				}
			}
			@Override
			public void cancel() {
			}
		});
		Tuple2WithTopicSchema schema = new Tuple2WithTopicSchema(env.getConfig());
		Properties props = new Properties();
		props.putAll(standardProps);
		props.putAll(secureProps);
		kafkaServer.produceIntoKafka(stream, "dummy", schema, props, null);
		env.execute("Write to topics");
		env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.getConfig().disableSysoutLogging();
		stream = env.addSource(kafkaServer.getConsumer(topics, schema, props));
		stream.flatMap(new FlatMapFunction<Tuple3<Integer, Integer, String>, Integer>() {
			Map<String, Integer> countPerTopic = new HashMap<>(numTopics);
			@Override
			public void flatMap(Tuple3<Integer, Integer, String> value, Collector<Integer> out) throws Exception {
				Integer count = countPerTopic.get(value.f2);
				if (count == null) {
					count = 1;
				} else {
					count++;
				}
				countPerTopic.put(value.f2, count);
				for (Map.Entry<String, Integer> el: countPerTopic.entrySet()) {
					if (el.getValue() < numElements) {
						break; 
					}
					if (el.getValue() > numElements) {
						throw new RuntimeException("There is a failure in the test. I've read " +
								el.getValue() + " from topic " + el.getKey());
					}
				}
				throw new SuccessException();
			}
		}).setParallelism(1);
		tryExecute(env, "Count elements from the topics");
		for (int i = 0; i < numTopics; i++) {
			final String topic = "topic-" + i;
			deleteTestTopic(topic);
		}
	}
public void runProduceConsumeMultipleTopics ( ) throws Exception { final int numTopics = 5 ; final int numElements = 20 ; StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env . getConfig ( ) . disableSysoutLogging ( ) ; final List < String > topics = new ArrayList < > ( ) ; for ( int i = 0 ; i < numTopics ; i ++ ) { final String topic = "topic-" + i ; topics . add ( topic ) ; createTestTopic ( topic , i + 1 /*partitions*/ , 1 ) ; } env . setParallelism ( 1 ) ; DataStream < Tuple3 < Integer , Integer , String > > stream = env . addSource ( new RichParallelSourceFunction < Tuple3 < Integer , Integer , String > > ( ) { @ Override public void run ( SourceContext < Tuple3 < Integer , Integer , String > > ctx ) throws Exception { int partition = getRuntimeContext ( ) . getIndexOfThisSubtask ( ) ; for ( int topicId = 0 ; topicId < numTopics ; topicId ++ ) { for ( int i = 0 ; i < numElements ; i ++ ) { ctx . collect ( new Tuple3 < > ( partition , i , "topic-" + topicId ) ) ; } } } @ Override public void cancel ( ) { } } ) ; Tuple2WithTopicSchema schema = new Tuple2WithTopicSchema ( env . getConfig ( ) ) ; Properties props = new Properties ( ) ; props . putAll ( standardProps ) ; props . putAll ( secureProps ) ; kafkaServer . produceIntoKafka ( stream , "dummy" , schema , props , null ) ; env . execute ( "Write to topics" ) ; env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env . getConfig ( ) . disableSysoutLogging ( ) ; stream = env . addSource ( kafkaServer . getConsumer ( topics , schema , props ) ) ; stream . flatMap ( new FlatMapFunction < Tuple3 < Integer , Integer , String > , Integer > ( ) { Map < String , Integer > countPerTopic = new HashMap < > ( numTopics ) ; @ Override public void flatMap ( Tuple3 < Integer , Integer , String > value , Collector < Integer > out ) throws Exception { Integer count = countPerTopic . get ( value . f2 ) ; if ( count == null ) { count = 1 ; } else { count ++ ; } countPerTopic . put ( value . f2 , count ) ; for ( Map . Entry < String , Integer > el : countPerTopic . entrySet ( ) ) { if ( el . getValue ( ) < numElements ) { break ; } if ( el . getValue ( ) > numElements ) { throw new RuntimeException ( "There is a failure in the test. I've read " + el . getValue ( ) + " from topic " + el . getKey ( ) ) ; } } throw new SuccessException ( ) ; } } ) . setParallelism ( 1 ) ; tryExecute ( env , "Count elements from the topics" ) ; for ( int i = 0 ; i < numTopics ; i ++ ) { final String topic = "topic-" + i ; deleteTestTopic ( topic ) ; } }
Test producing and consuming into multiple topics.
Test producing and consuming into multiple topics .
// create topics with content // create topic // before FLINK-6078 the RemoteExecutionEnvironment set the parallelism to 1 as well // run first job, producing into all topics // run second job consuming from multiple topics // check map: // not enough yet // we've seen messages from all topics // delete all topics again