@Test public void testListener() throws Exception { time.allowSleepCalls(0); DataNodeConfigChangeListener listener = mock(DataNodeConfigChangeListener.class); compositeSource.addDataNodeConfigChangeListener(listener); DataNodeConfigChangeListener primaryListener = captureRegisteredListener(primarySource); DataNodeConfigChangeListener secondaryListener = captureRegisteredListener(secondarySource); DataNodeConfig configOne = createConfig(1, 1); primaryListener.onDataNodeConfigChange(Collections.singleton(configOne)); verify(listener).onDataNodeConfigChange(Collections.singleton(configOne)); time.allowSleepCalls(1); assertTrue("Expected 0 inconsistencies so far", TestUtils.checkAndSleep(0L, metrics.listenerInconsistencyCount::getCount, 500)); assertTrue("Expected 1 transient inconsistency so far", TestUtils.checkAndSleep(1L, metrics.listenerTransientInconsistencyCount::getCount, 500)); time.allowSleepCalls(2); assertTrue("Expected 1 inconsistency so far", TestUtils.checkAndSleep(1L, metrics.listenerInconsistencyCount::getCount, 500)); assertTrue("Expected 2 transient inconsistencies so far", TestUtils.checkAndSleep(2L, metrics.listenerTransientInconsistencyCount::getCount, 500)); secondaryListener.onDataNodeConfigChange(Collections.singleton(configOne)); time.allowSleepCalls(1); assertEquals("Expected no more inconsistencies", 1, metrics.listenerInconsistencyCount.getCount()); assertEquals("Expected no more transient inconsistencies", 2, metrics.listenerTransientInconsistencyCount.getCount()); DataNodeConfigChangeListener newListener = mock(DataNodeConfigChangeListener.class); compositeSource.addDataNodeConfigChangeListener(newListener); verify(primarySource).addDataNodeConfigChangeListener(newListener); verify(secondarySource, never()).addDataNodeConfigChangeListener(newListener); }
@Override public void addDataNodeConfigChangeListener(DataNodeConfigChangeListener listener) throws Exception { if (recordingListenersRegistered.compareAndSet(false, true)) { ConfigChangeRecorder primaryListener = new ConfigChangeRecorder(listener); ConfigChangeRecorder secondaryListener = new ConfigChangeRecorder(null); primarySource.addDataNodeConfigChangeListener(primaryListener); try { secondarySource.addDataNodeConfigChangeListener(secondaryListener); Utils.newThread("DataNodeConfigConsistencyChecker", new ConsistencyChecker(time, metrics, primaryListener, secondaryListener), false).start(); } catch (Exception e) { metrics.addListenerSwallowedErrorCount.inc(); LOGGER.error("Error from secondarySource.addDataNodeConfigChangeListener({})", listener, e); } } else { primarySource.addDataNodeConfigChangeListener(listener); } }
CompositeDataNodeConfigSource implements DataNodeConfigSource { @Override public void addDataNodeConfigChangeListener(DataNodeConfigChangeListener listener) throws Exception { if (recordingListenersRegistered.compareAndSet(false, true)) { ConfigChangeRecorder primaryListener = new ConfigChangeRecorder(listener); ConfigChangeRecorder secondaryListener = new ConfigChangeRecorder(null); primarySource.addDataNodeConfigChangeListener(primaryListener); try { secondarySource.addDataNodeConfigChangeListener(secondaryListener); Utils.newThread("DataNodeConfigConsistencyChecker", new ConsistencyChecker(time, metrics, primaryListener, secondaryListener), false).start(); } catch (Exception e) { metrics.addListenerSwallowedErrorCount.inc(); LOGGER.error("Error from secondarySource.addDataNodeConfigChangeListener({})", listener, e); } } else { primarySource.addDataNodeConfigChangeListener(listener); } } }
CompositeDataNodeConfigSource implements DataNodeConfigSource { @Override public void addDataNodeConfigChangeListener(DataNodeConfigChangeListener listener) throws Exception { if (recordingListenersRegistered.compareAndSet(false, true)) { ConfigChangeRecorder primaryListener = new ConfigChangeRecorder(listener); ConfigChangeRecorder secondaryListener = new ConfigChangeRecorder(null); primarySource.addDataNodeConfigChangeListener(primaryListener); try { secondarySource.addDataNodeConfigChangeListener(secondaryListener); Utils.newThread("DataNodeConfigConsistencyChecker", new ConsistencyChecker(time, metrics, primaryListener, secondaryListener), false).start(); } catch (Exception e) { metrics.addListenerSwallowedErrorCount.inc(); LOGGER.error("Error from secondarySource.addDataNodeConfigChangeListener({})", listener, e); } } else { primarySource.addDataNodeConfigChangeListener(listener); } } CompositeDataNodeConfigSource(DataNodeConfigSource primarySource, DataNodeConfigSource secondarySource,
      Time time, DataNodeConfigSourceMetrics metrics); }
CompositeDataNodeConfigSource implements DataNodeConfigSource { @Override public void addDataNodeConfigChangeListener(DataNodeConfigChangeListener listener) throws Exception { if (recordingListenersRegistered.compareAndSet(false, true)) { ConfigChangeRecorder primaryListener = new ConfigChangeRecorder(listener); ConfigChangeRecorder secondaryListener = new ConfigChangeRecorder(null); primarySource.addDataNodeConfigChangeListener(primaryListener); try { secondarySource.addDataNodeConfigChangeListener(secondaryListener); Utils.newThread("DataNodeConfigConsistencyChecker", new ConsistencyChecker(time, metrics, primaryListener, secondaryListener), false).start(); } catch (Exception e) { metrics.addListenerSwallowedErrorCount.inc(); LOGGER.error("Error from secondarySource.addDataNodeConfigChangeListener({})", listener, e); } } else { primarySource.addDataNodeConfigChangeListener(listener); } } CompositeDataNodeConfigSource(DataNodeConfigSource primarySource, DataNodeConfigSource secondarySource,
      Time time, DataNodeConfigSourceMetrics metrics); @Override void addDataNodeConfigChangeListener(DataNodeConfigChangeListener listener); @Override boolean set(DataNodeConfig config); @Override DataNodeConfig get(String instanceName); @Override void close(); }
CompositeDataNodeConfigSource implements DataNodeConfigSource { @Override public void addDataNodeConfigChangeListener(DataNodeConfigChangeListener listener) throws Exception { if (recordingListenersRegistered.compareAndSet(false, true)) { ConfigChangeRecorder primaryListener = new ConfigChangeRecorder(listener); ConfigChangeRecorder secondaryListener = new ConfigChangeRecorder(null); primarySource.addDataNodeConfigChangeListener(primaryListener); try { secondarySource.addDataNodeConfigChangeListener(secondaryListener); Utils.newThread("DataNodeConfigConsistencyChecker", new ConsistencyChecker(time, metrics, primaryListener, secondaryListener), false).start(); } catch (Exception e) { metrics.addListenerSwallowedErrorCount.inc(); LOGGER.error("Error from secondarySource.addDataNodeConfigChangeListener({})", listener, e); } } else { primarySource.addDataNodeConfigChangeListener(listener); } } CompositeDataNodeConfigSource(DataNodeConfigSource primarySource, DataNodeConfigSource secondarySource,
      Time time, DataNodeConfigSourceMetrics metrics); @Override void addDataNodeConfigChangeListener(DataNodeConfigChangeListener listener); @Override boolean set(DataNodeConfig config); @Override DataNodeConfig get(String instanceName); @Override void close();  }