public List<ConsumerRecord<String, String>> consumeMessages(String topic, int threshold) {
    int emptyPollCount = 0;
    List<ConsumerRecord<String, String>> recordList = new ArrayList<>();
    KafkaConsumer kafkaConsumer =
        topicConsumerMap.computeIfAbsent(topic, t -> new KafkaConsumer<>(createConsumerConfigs()));
    kafkaConsumer.subscribe(Collections.singletonList(topic));
    while (emptyPollCount < numEmptyPolls && recordList.size() < threshold) {
      ConsumerRecords<String, String> records = kafkaConsumer.poll(POLL_TIMEOUT_MS);
      LOG.info("Read {} messages from topic: {}", records.count(), topic);
      if (!records.isEmpty()) {
        Iterator<ConsumerRecord<String, String>> iterator = records.iterator();
        while (iterator.hasNext() && recordList.size() < threshold) {
          ConsumerRecord record = iterator.next();
          LOG.info("Read key: {} val: {} from topic: {} on partition: {}", record.key(), record.value(), record.topic(),
              record.partition());
          recordList.add(record);
          emptyPollCount = 0;
        }
      } else {
        emptyPollCount++;
      }
    }
    return recordList;
  }
Read messages from the provided list of topics until {@param threshold} messages have been read or until
{@link #numEmptyPolls} polls return no messages.

The default poll time out is determined by {@link #POLL_TIMEOUT_MS} and the number of empty polls are
determined by {@link #numEmptyPolls}

@param topic the topic to consume from
@param threshold the number of messages to consume
@return the list of {@link ConsumerRecord}s whose size can be atmost {@param threshold}
samza-test/src/test/java/org/apache/samza/test/framework/StreamApplicationIntegrationTestHarness.java