MysqlSource extends AbstractDataStoreSource<BinlogEvent> { public void commitCheckpoint(final Mutation<?> mutation) { final MysqlSourceState savedState = lastSavedState.get(); if (mutation == null || savedState == null) { return; } Preconditions.checkState(mutation instanceof MysqlMutation); final MysqlMutationMetadata metadata = ((MysqlMutation) mutation).getMetadata(); BinlogFilePos mutationPosition = metadata.getFilePos(); BinlogFilePos savedStatePosition = savedState.getLastPosition(); if ((BinlogFilePos.shouldCompareUsingFilePosition(mutationPosition, savedStatePosition) && savedState.getLastOffset() >= metadata.getId()) || (mutationPosition.getGtidSet() != null && mutationPosition.getGtidSet().isContainedWithin(savedStatePosition.getGtidSet()))) { return; } final MysqlSourceState newState = new MysqlSourceState( metadata.getTimestamp(), metadata.getId(), currentLeaderEpoch.get(), metadata.getLastTransaction().getPosition()); saveState(newState); stateHistory.add(newState); stateRollbackCount.set(1); } 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 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()); }