@Test
public void testResetToLastValidState() throws Exception {

  StateHistory<MysqlSourceState> stateHistory = createTestStateHistory();
  TestSource source = new TestSource(stateHistory);
  MysqlSourceState savedState = mock(MysqlSourceState.class);
  MysqlSourceState earliestState =
      new MysqlSourceState(0L, 0L, 0L, MysqlSource.EARLIEST_BINLOG_POS);
  when(stateRepository.read()).thenReturn(savedState);
  MysqlSourceState firstState = mock(MysqlSourceState.class);
  MysqlSourceState secondState = mock(MysqlSourceState.class);
  MysqlSourceState thirdState = mock(MysqlSourceState.class);
  MysqlSourceState fourthState = mock(MysqlSourceState.class);
  stateHistory.add(firstState);
  stateHistory.add(secondState);
  stateHistory.add(thirdState);
  source.initialize();
  source.resetToLastValidState();
  assertEquals(thirdState, source.getLastSavedState().get());
  source.resetToLastValidState();
  assertEquals(firstState, source.getLastSavedState().get());
  assertTrue(stateHistory.isEmpty());
  source.resetToLastValidState();
  assertEquals(earliestState, source.getLastSavedState().get());
  stateHistory.add(firstState);
  stateHistory.add(secondState);
  stateHistory.add(thirdState);
  stateHistory.add(fourthState);
  source.resetToLastValidState();
  assertEquals(firstState, source.getLastSavedState().get());
  stateHistory.add(firstState);
  stateHistory.add(secondState);
  source.resetToLastValidState();
  assertEquals(earliestState, source.getLastSavedState().get());
  assertTrue(stateHistory.isEmpty());
  BinlogFilePos filePos = new BinlogFilePos("mysql-binlog.123450", 18, 156);
  Transaction lastTransaction = new Transaction(0L, 0L, filePos);
  MysqlMutationMetadata metadata =
      new MysqlMutationMetadata(null, filePos, null, 0L, 1L, 23L, null, lastTransaction, 0L, 0);
  source.checkpoint(new MysqlInsertMutation(metadata, null));
  assertFalse(stateHistory.isEmpty());
  source.resetToLastValidState();
  assertEquals(new MysqlSourceState(23L, 1L, 0L, filePos), source.getLastSavedState().get());
}