@Override
    public void run() {
      long lastSuccessTimeMs = time.milliseconds();
      while (!Thread.currentThread().isInterrupted()) {
        try {
          time.sleep(CHECK_PERIOD_MS);
        } catch (InterruptedException e) {
          LOGGER.warn("ConsistencyChecker sleep interrupted, shutting down");
          Thread.currentThread().interrupt();
          break;
        }
        long currentTimeMs = time.milliseconds();
        Map<String, DataNodeConfig> primaryConfigsSeen = primaryListener.configsSeen;
        Map<String, DataNodeConfig> secondaryConfigsSeen = secondaryListener.configsSeen;
        if (primaryConfigsSeen.equals(secondaryConfigsSeen)) {
          lastSuccessTimeMs = currentTimeMs;
          LOGGER.debug("Sources consistent at {}", currentTimeMs);
        } else if ((currentTimeMs - CHECK_PERIOD_MS * MAX_PERIODS_WITH_INCONSISTENCY) >= lastSuccessTimeMs) {
          metrics.listenerInconsistencyCount.inc();
          LOGGER.warn("Inconsistency detected for multiple periods. primary={}, secondary={}", primaryConfigsSeen,
              secondaryConfigsSeen);
        } else {
          metrics.listenerTransientInconsistencyCount.inc();
          LOGGER.debug("Inconsistency detected at {}, waiting to see if state converges. primary={}, secondary={}",
              currentTimeMs, primaryConfigsSeen, secondaryConfigsSeen);
        }
      }
    }
@param time {@link Time} instance.
@param metrics {@link DataNodeConfigSourceMetrics} instance.
@param primaryListener the first {@link ConfigChangeRecorder} to compare.
@param secondaryListener the second {@link ConfigChangeRecorder} to compare.
ambry-clustermap/src/main/java/com/github/ambry/clustermap/CompositeDataNodeConfigSource.java