MysqlSource extends AbstractDataStoreSource<BinlogEvent> { protected void onCommunicationError(final Exception ex) { metrics.communicationFailure(ex); if (ex instanceof InvalidBinlogPositionException) { resetToLastValidState(); } throw new RuntimeException(ex); } 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 testOnCommunicationError() throws Exception { TestSource source = new TestSource(); source.addListener(listener); source.setPosition(null); try { source.onCommunicationError(new RuntimeException()); fail("Should not reach here"); } catch (Exception ex) { } assertNull(source.getLastSavedState().get()); try { source.onCommunicationError(new InvalidBinlogPositionException("")); fail("Should not reach here"); } catch (Exception ex) { } assertEquals( MysqlSource.EARLIEST_BINLOG_POS, source.getLastSavedState().get().getLastPosition()); }