MysqlSource extends AbstractDataStoreSource<BinlogEvent> { void resetToLastValidState() { if (stateHistory.size() >= stateRollbackCount.get()) { final MysqlSourceState newState = stateHistory.removeLast(stateRollbackCount.get()); saveState(newState); metrics.resetSourcePosition(); log.info("Reset source {} position to {}.", name, newState.getLastPosition()); stateRollbackCount.accumulateAndGet( STATE_ROLLBACK_BACKOFF_RATE, (value, rate) -> value * rate); } else { stateHistory.clear(); saveState(getEarliestState()); metrics.resetEarliestPosition(); log.info("Reset source {} position to earliest.", name); } } 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 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()); }