@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)); when(nextState.getCurrentLeaderEpoch()).thenReturn(4l); stateRepository.save(nextState); assertEquals(state, updatedState.get()); when(nextState.getCurrentLeaderEpoch()).thenReturn(5l); stateRepository.save(nextState); assertEquals(nextState, updatedState.get()); when(nextState.getCurrentLeaderEpoch()).thenReturn(6l); stateRepository.save(nextState); assertEquals(nextState, updatedState.get()); }
public void save(@NonNull final S state) { try { repository.update( state, (currentValue, nextValue) -> { if (currentValue.getCurrentLeaderEpoch() > nextValue.getCurrentLeaderEpoch()) { log.warn("Will not update mysql state: current={}, next={}", currentValue, nextValue); return currentValue; } return nextValue; }); } catch (Exception ex) { log.error("Failed to save state for source " + sourceName, ex); metrics.stateSaveFailure(ex); throw new RuntimeException(ex); } log.info("Saved state for source {}. state={}", sourceName, state); metrics.stateSave(); }
StateRepository { public void save(@NonNull final S state) { try { repository.update( state, (currentValue, nextValue) -> { if (currentValue.getCurrentLeaderEpoch() > nextValue.getCurrentLeaderEpoch()) { log.warn("Will not update mysql state: current={}, next={}", currentValue, nextValue); return currentValue; } return nextValue; }); } catch (Exception ex) { log.error("Failed to save state for source " + sourceName, ex); metrics.stateSaveFailure(ex); throw new RuntimeException(ex); } log.info("Saved state for source {}. state={}", sourceName, state); metrics.stateSave(); } }
StateRepository { public void save(@NonNull final S state) { try { repository.update( state, (currentValue, nextValue) -> { if (currentValue.getCurrentLeaderEpoch() > nextValue.getCurrentLeaderEpoch()) { log.warn("Will not update mysql state: current={}, next={}", currentValue, nextValue); return currentValue; } return nextValue; }); } catch (Exception ex) { log.error("Failed to save state for source " + sourceName, ex); metrics.stateSaveFailure(ex); throw new RuntimeException(ex); } log.info("Saved state for source {}. state={}", sourceName, state); metrics.stateSave(); }  }
StateRepository { public void save(@NonNull final S state) { try { repository.update( state, (currentValue, nextValue) -> { if (currentValue.getCurrentLeaderEpoch() > nextValue.getCurrentLeaderEpoch()) { log.warn("Will not update mysql state: current={}, next={}", currentValue, nextValue); return currentValue; } return nextValue; }); } catch (Exception ex) { log.error("Failed to save state for source " + sourceName, ex); metrics.stateSaveFailure(ex); throw new RuntimeException(ex); } log.info("Saved state for source {}. state={}", sourceName, state); metrics.stateSave(); }  void save(@NonNull final S state); S read(); }
StateRepository { public void save(@NonNull final S state) { try { repository.update( state, (currentValue, nextValue) -> { if (currentValue.getCurrentLeaderEpoch() > nextValue.getCurrentLeaderEpoch()) { log.warn("Will not update mysql state: current={}, next={}", currentValue, nextValue); return currentValue; } return nextValue; }); } catch (Exception ex) { log.error("Failed to save state for source " + sourceName, ex); metrics.stateSaveFailure(ex); throw new RuntimeException(ex); } log.info("Saved state for source {}. state={}", sourceName, state); metrics.stateSave(); }  void save(@NonNull final S state); S read();  }