@Test(expected = RuntimeException.class) public void testSaveFailure() throws Exception { doThrow(new RuntimeException()) .when(repository) .update(any(MysqlSourceState.class), any(Repository.DataUpdater.class)); try { stateRepository.save(mock(MysqlSourceState.class)); } catch (RuntimeException ex) { verify(metrics, times(1)).stateSaveFailure(any(Exception.class)); throw ex; } }
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();  }