@Test
public void testSave() throws Exception {

  MysqlSourceState state = mock(MysqlSourceState.class);
  MysqlSourceState nextState = mock(MysqlSourceState.class);
  AtomicReference<MysqlSourceState> updatedState = new AtomicReference<>();
  when(state.getCurrentLeaderEpoch()).thenReturn(5l);
  doAnswer(
          new Answer<MysqlSourceState>() {
            @Override
            public MysqlSourceState answer(InvocationOnMock invocation) throws Throwable {
              Object[] args = invocation.getArguments();
              MysqlSourceState newState = (MysqlSourceState) args[0];
              Repository.DataUpdater<MysqlSourceState> updater =
                  (Repository.DataUpdater<MysqlSourceState>) args[1];
              updatedState.set(updater.apply(state, newState));
              return null;
            }
          })
      .when(repository)
      .update(any(MysqlSourceState.class), any(Repository.DataUpdater.class));
  // Test new leader epoch leader less than current
  when(nextState.getCurrentLeaderEpoch()).thenReturn(4l);
  stateRepository.save(nextState);
  assertEquals(state, updatedState.get());
  // Test new leader epoch leader same as current
  when(nextState.getCurrentLeaderEpoch()).thenReturn(5l);
  stateRepository.save(nextState);
  assertEquals(nextState, updatedState.get());
  // Test new leader epoch leader greather current
  when(nextState.getCurrentLeaderEpoch()).thenReturn(6l);
  stateRepository.save(nextState);
  assertEquals(nextState, updatedState.get());
}