MysqlSource extends AbstractDataStoreSource<BinlogEvent> { void saveState(@NonNull final MysqlSourceState state) { stateRepository.save(state); lastSavedState.set(state); } MysqlSource(
      @NonNull final String name,
      @NonNull final DataSource dataSource,
      @NonNull final Set<String> tableNames,
      @NonNull final TableCache tableCache,
      @NonNull final StateRepository<MysqlSourceState> stateRepository,
      @NonNull final StateHistory<MysqlSourceState> stateHistory,
      @NonNull final BinlogFilePos initialBinlogFilePosition,
      @NonNull final MysqlSchemaManager schemaManager,
      @NonNull final MysqlSourceMetrics metrics,
      @NonNull final AtomicLong currentLeaderEpoch,
      @NonNull final AtomicReference<Transaction> lastTransaction,
      @NonNull final AtomicReference<MysqlSourceState> lastSavedState); abstract void setPosition(BinlogFilePos pos); void commitCheckpoint(final Mutation<?> mutation); static final BinlogFilePos LATEST_BINLOG_POS; static final BinlogFilePos EARLIEST_BINLOG_POS; }
@Test public void testSaveState() throws Exception { TestSource source = new TestSource(); MysqlSourceState savedState = mock(MysqlSourceState.class); MysqlSourceState newState = mock(MysqlSourceState.class); when(stateRepository.read()).thenReturn(savedState); source.saveState(newState); verify(stateRepository, times(1)).save(newState); assertEquals(newState, source.getLastSavedState().get()); }