JobRuntimePredictionSelectors { public static CloseableReference<JobRuntimePredictionSelector> reloadedOnConfigurationUpdate(Config config, Function<Config, JobRuntimePredictionSelector> factory) { AtomicReference<JobRuntimePredictionSelector> last = new AtomicReference<>(factory.apply(config)); ConfigListener listener = new ConfigListener() { @Override public void onConfigAdded(Config config) { tryReload(config); } @Override public void onConfigRemoved(Config config) { tryReload(config); } @Override public void onConfigUpdated(Config config) { tryReload(config); } @Override public void onError(Throwable error, Config config) { tryReload(config); } private void tryReload(Config updated) { try { last.set(factory.apply(config)); } catch (Exception e) { logger.info("Failed to reload job runtime prediction selectors", e); } } }; config.addListener(listener); JobRuntimePredictionSelector wrapper = (jobDescriptor, predictions) -> last.get().apply(jobDescriptor, predictions); return CloseableReference.<JobRuntimePredictionSelector>newBuilder() .withResource(wrapper) .withCloseAction(() -> config.removeListener(listener)) .withSwallowException(true) .withSerialize(true) .build(); } private  JobRuntimePredictionSelectors(); static JobRuntimePredictionSelector never(); static JobRuntimePredictionSelector best(); static JobRuntimePredictionSelector best(Map<String, String> selectionMetadata); static JobRuntimePredictionSelector aboveThreshold(double runtimeThresholdInSeconds,
                                                              double sigmaThreshold,
                                                              double quantile,
                                                              Map<String, String> selectionMetadata); static JobRuntimePredictionSelector aboveThreshold(Config config, Map<String, String> selectionMetadata); static Map<String, JobRuntimePredictionSelector> aboveThresholds(Config config, Map<String, String> selectionMetadata); static CloseableReference<JobRuntimePredictionSelector> reloadedOnConfigurationUpdate(Config config, Function<Config, JobRuntimePredictionSelector> factory); static JobRuntimePredictionSelector abTestRandom(Map<String, JobRuntimePredictionSelector> selectors);  }
@Test public void testReloadedOnConfigurationUpdate() { SettableConfig config = new DefaultSettableConfig(); config.setProperty("root.runtimeThresholdInSeconds", "60"); config.setProperty("root.sigmaThreshold", "10"); CloseableReference<JobRuntimePredictionSelector> selectorRef = reloadedOnConfigurationUpdate( config.getPrefixedView("root"), configUpdate -> aboveThreshold(configUpdate, METADATA) ); JobRuntimePredictionSelector selector = selectorRef.get(); checkSelection(selector.apply(JOB_DESCRIPTOR, JOB_PREDICTIONS), HIGH_QUANTILE); config.setProperty("root.runtimeThresholdInSeconds", "6"); assertThat(selector.apply(JOB_DESCRIPTOR, JOB_PREDICTIONS)).isEmpty(); selectorRef.close(); }