@Override
    public void describe(TaskApplicationDescriptor appDescriptor) {
      KafkaSystemDescriptor sd = new KafkaSystemDescriptor(SYSTEM);
      KafkaInputDescriptor<String> isd = sd.getInputDescriptor(INPUT_STREAM, new StringSerde());
      appDescriptor.withInputStream(isd)
          .withTaskFactory((StreamTaskFactory) () -> new CheckpointTask(this.handleIntermediateShutdown));
    }
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