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
/**
	 * Read topic to list, only using Kafka code.
	 */
Read topic to list, only using Kafka code.
Read topic to list only using Kafka code .
private static List<MessageAndMetadata<byte[], byte[]>> readTopicToList(String topicName, ConsumerConfig config, final int stopAfter) {
		ConsumerConnector consumerConnector = Consumer.createJavaConsumerConnector(config);
		Map<String, Integer> topicCountMap = Collections.singletonMap(topicName, 1);
		Map<String, List<KafkaStream<byte[], byte[]>>> streams = consumerConnector.createMessageStreams(topicCountMap);
		if (streams.size() != 1) {
			throw new RuntimeException("Expected only one message stream but got " + streams.size());
		}
		List<KafkaStream<byte[], byte[]>> kafkaStreams = streams.get(topicName);
		if (kafkaStreams == null) {
			throw new RuntimeException("Requested stream not available. Available streams: " + streams.toString());
		}
		if (kafkaStreams.size() != 1) {
			throw new RuntimeException("Requested 1 stream from Kafka, bot got " + kafkaStreams.size() + " streams");
		}
		LOG.info("Opening Consumer instance for topic '{}' on group '{}'", topicName, config.groupId());
		ConsumerIterator<byte[], byte[]> iteratorToRead = kafkaStreams.get(0).iterator();
		List<MessageAndMetadata<byte[], byte[]>> result = new ArrayList<>();
		int read = 0;
		while (iteratorToRead.hasNext()) {
			read++;
			result.add(iteratorToRead.next());
			if (read == stopAfter) {
				LOG.info("Read " + read + " elements");
				return result;
			}
		}
		return result;
	}
private static List < MessageAndMetadata < byte [ ] , byte [ ] > > readTopicToList ( String topicName , ConsumerConfig config , final int stopAfter ) { ConsumerConnector consumerConnector = Consumer . createJavaConsumerConnector ( config ) ; Map < String , Integer > topicCountMap = Collections . singletonMap ( topicName , 1 ) ; Map < String , List < KafkaStream < byte [ ] , byte [ ] > > > streams = consumerConnector . createMessageStreams ( topicCountMap ) ; if ( streams . size ( ) != 1 ) { throw new RuntimeException ( "Expected only one message stream but got " + streams . size ( ) ) ; } List < KafkaStream < byte [ ] , byte [ ] > > kafkaStreams = streams . get ( topicName ) ; if ( kafkaStreams == null ) { throw new RuntimeException ( "Requested stream not available. Available streams: " + streams . toString ( ) ) ; } if ( kafkaStreams . size ( ) != 1 ) { throw new RuntimeException ( "Requested 1 stream from Kafka, bot got " + kafkaStreams . size ( ) + " streams" ) ; } LOG . info ( "Opening Consumer instance for topic '{}' on group '{}'" , topicName , config . groupId ( ) ) ; ConsumerIterator < byte [ ] , byte [ ] > iteratorToRead = kafkaStreams . get ( 0 ) . iterator ( ) ; List < MessageAndMetadata < byte [ ] , byte [ ] > > result = new ArrayList < > ( ) ; int read = 0 ; while ( iteratorToRead . hasNext ( ) ) { read ++ ; result . add ( iteratorToRead . next ( ) ) ; if ( read == stopAfter ) { LOG . info ( "Read " + read + " elements" ) ; return result ; } } return result ; }
Read topic to list, only using Kafka code.
Read topic to list only using Kafka code .
// we request only one stream per consumer instance. Kafka will make sure that each consumer group // will see each message only once.