@Override
  public void finalizeCheckpoint() {
    try {
      checkState(committer.isPresent());
      committer.get().get().commitOffset(offset);
    } catch (Exception e) {
      logger.warn("Failed to finalize checkpoint.", e);
    }
  }
@Test
  public void testFinalize() throws Exception {
    mark.finalizeCheckpoint();
    verify(committer).commitOffset(OFFSET);
  }