class 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 {
                    // Use the past config, in case it is prefixed view
                    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();

}