OffsetCheckpointMark implements CheckpointMark { @Override public void finalizeCheckpoint() throws IOException { if (!finalizer.isPresent()) { return; } try { finalizer.get().finalizeOffsets(partitionOffsetMap); } catch (StatusException e) { throw new IOException(e); } } OffsetCheckpointMark(OffsetFinalizer finalizer, Map<Partition, Offset> partitionOffsetMap); private  OffsetCheckpointMark(Map<Long, Long> encodedMap); @Override void finalizeCheckpoint();  }
@Test public void finalizeFinalizesWithOffsets() throws Exception { Map<Partition, Offset> map = ImmutableMap.of(Partition.of(10), Offset.of(15), Partition.of(85), Offset.of(0)); OffsetFinalizer finalizer = mock(OffsetFinalizer.class); OffsetCheckpointMark mark = new OffsetCheckpointMark(finalizer, map); mark.finalizeCheckpoint(); verify(finalizer).finalizeOffsets(mapCaptor.capture()); assertThat(mapCaptor.getValue(), equalTo(map)); }