@SuppressWarnings("unchecked")
  private void verifyNumMessages(KafkaConsumer consumer, String topic, int expectedNumMessages) {
    consumer.subscribe(Collections.singletonList(topic));

    int count = 0;
    int emptyPollCount = 0;

    while (count < expectedNumMessages && emptyPollCount < 5) {
      ConsumerRecords records = consumer.poll(5000);
      if (!records.isEmpty()) {
        for (ConsumerRecord record : (Iterable<ConsumerRecord>) records) {
          Assert.assertEquals(new String((byte[]) record.value()), String.valueOf(count));
          count++;
        }
      } else {
        emptyPollCount++;
      }
    }

    Assert.assertEquals(count, expectedNumMessages);
  }
Consumes data from the topic until there are no new messages for a while
and asserts that the number of consumed messages is as expected.
samza-test/src/test/java/org/apache/samza/test/processor/TestStreamProcessor.java