public static <K,V>List<KeyValue<K,V>> readKeyValues(String topic,Properties consumerConfig,int maxMessages){
  KafkaConsumer<K,V> consumer=new KafkaConsumer<>(consumerConfig);
  consumer.subscribe(Collections.singletonList(topic));
  int pollIntervalMs=100;
  int maxTotalPollTimeMs=2000;
  int totalPollTimeMs=0;
  List<KeyValue<K,V>> consumedValues=new ArrayList<>();
  while (totalPollTimeMs < maxTotalPollTimeMs && continueConsuming(consumedValues.size(),maxMessages)) {
    totalPollTimeMs+=pollIntervalMs;
    ConsumerRecords<K,V> records=consumer.poll(pollIntervalMs);
    for (    ConsumerRecord<K,V> record : records) {
      consumedValues.add(new KeyValue<>(record.key(),record.value()));
    }
  }
  consumer.close();
  return consumedValues;
}

public static < K , V > List < KeyValue < K , V > > readKeyValues ( String topic , Properties consumerConfig , int maxMessages ) { KafkaConsumer < K , V > consumer = new KafkaConsumer < > ( consumerConfig ) ; consumer . subscribe ( Collections . singletonList ( topic ) ) ; int pollIntervalMs = 100 ; int maxTotalPollTimeMs = 2000 ; int totalPollTimeMs = 0 ; List < KeyValue < K , V > > consumedValues = new ArrayList < > ( ) ; while ( totalPollTimeMs < maxTotalPollTimeMs && continueConsuming ( consumedValues . size ( ) , maxMessages ) ) { totalPollTimeMs += pollIntervalMs ; ConsumerRecords < K , V > records = consumer . poll ( pollIntervalMs ) ; for ( ConsumerRecord < K , V > record : records ) { consumedValues . add ( new KeyValue < > ( record . key ( ) , record . value ( ) ) ) ; } } consumer . close ( ) ; return consumedValues ; }
returns up to ` maxmessages ` by reading via the provided consumer ( the topic ( s ) to read from are already configured in the consumer ) .
