private static void verifyProcessedMessagesSecondRun() {
    assertEquals(6, PROCESSED.size());
    assertEquals(1, PROCESSED.get(commitMessage(0, 0)).get());
    assertEquals(2, PROCESSED.get(noCommitMessage(0, 1)).get());
    assertEquals(1, PROCESSED.get(commitMessage(0, 2)).get());
    assertEquals(1, PROCESSED.get(commitMessage(1, 0)).get());
    assertEquals(2, PROCESSED.get(noCommitMessage(1, 1)).get());
    assertEquals(1, PROCESSED.get(commitMessage(1, 2)).get());
  }
For each partition: re-process the second message (for 2 total of the second message), receive the third message.
samza-test/src/test/java/org/apache/samza/test/kafka/KafkaCheckpointManagerIntegrationTest.java