protected void waitUntilMessagesLeftN(int untilLeft) {
    int attempts = ATTEMPTS_NUMBER;
    while (attempts > 0) {
      long leftEventsCount = TestZkStreamProcessorBase.TestStreamTask.endLatch.getCount();
      //System.out.println("2current count = " + leftEventsCount);
      if (leftEventsCount == untilLeft) { // that much should be left
        System.out.println("2read all. current count = " + leftEventsCount);
        break;
      }
      TestZkUtils.sleepMs(5000);
      attempts--;
    }
    Assert.assertTrue("Didn't read all the leftover events in " + ATTEMPTS_NUMBER + " attempts", attempts > 0);
  }
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