protected void verifyNumMessages(String topic, final Map<Integer, Boolean> expectedValues, int expectedNumMessages) {
    consumer.subscribe(Collections.singletonList(topic));

    Map<Integer, Boolean> map = new HashMap<>(expectedValues);
    int count = 0;
    int emptyPollCount = 0;

    while (count < expectedNumMessages && emptyPollCount < 5) {
      ConsumerRecords records = consumer.poll(5000);
      if (!records.isEmpty()) {
        Iterator<ConsumerRecord> iterator = records.iterator();
        while (iterator.hasNext()) {
          ConsumerRecord record = iterator.next();
          String val = new String((byte[]) record.value());
          LOG.info("Got value " + val + "; count = " + count + "; out of " + expectedNumMessages);
          Integer valI = Integer.valueOf(val);
          if (valI < BAD_MESSAGE_KEY) {
            map.put(valI, true);
            count++;
          }
        }
      } else {
        emptyPollCount++;
        LOG.warn("empty polls " + emptyPollCount);
      }
    }
    // filter out numbers we did not get
    long numFalse = map.values().stream().filter(v -> !v).count();
    Assert.assertEquals("didn't get this number of events ", 0, numFalse);
    Assert.assertEquals(expectedNumMessages, count);
  }
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/processor/TestZkStreamProcessorBase.java