@Test
public void testCommitCheckpoint() throws Exception {

  StateHistory<MysqlSourceState> stateHistory = createTestStateHistory();
  TestSource source = new TestSource(stateHistory);
  Row row = new Row(null, ImmutableMap.of());
  BinlogFilePos filePos = new BinlogFilePos("mysql-binlog.123450", 18, 156);
  Transaction lastTransaction = new Transaction(0L, 0L, filePos);
  MysqlMutationMetadata metadata =
      new MysqlMutationMetadata(null, filePos, null, 0L, 0L, 0L, null, lastTransaction, 0, 0);
  MysqlMutation mutation = new MysqlInsertMutation(metadata, row);
  MysqlSourceState savedState =
      new MysqlSourceState(SAVED_TIMESTAMP, SAVED_OFFSET, 0L, BINLOG_FILE_POS);
  when(stateRepository.read()).thenReturn(savedState);
  source.initialize();
  source.checkpoint(mutation);
  assertEquals(savedState, source.getLastSavedState().get());
  source.checkpoint(null);
  assertEquals(savedState, source.getLastSavedState().get());
  long newOffset = SAVED_OFFSET + 1;
  metadata =
      new MysqlMutationMetadata(
          null, filePos, null, 0L, newOffset, 23L, null, lastTransaction, 0, 0);
  mutation = new MysqlInsertMutation(metadata, row);
  source.checkpoint(mutation);
  assertEquals(
      new MysqlSourceState(23L, newOffset, 0L, filePos), source.getLastSavedState().get());
  assertEquals(stateHistory.removeLast(), source.getLastSavedState().get());
}