@Test public void testWorkerHealthMetrics_healthStatusExpired() throws IOException { final FakeTicker ticker = new FakeTicker(); final int healthStatusExpiry = 1000; CacheConfig.setValidationEnabled(conf, true); CacheConfig.setHealthStatusExpiry(conf, healthStatusExpiry); try (BookKeeperMetrics bookKeeperMetrics = new BookKeeperMetrics(conf, metrics)) { final CoordinatorBookKeeper coordinatorBookKeeper = new CoordinatorBookKeeper(conf, bookKeeperMetrics, ticker); final Gauge liveWorkerGauge = metrics.getGauges().get(BookKeeperMetrics.HealthMetric.LIVE_WORKER_GAUGE.getMetricName()); final Gauge cachingValidatedWorkerGauge = metrics.getGauges().get(BookKeeperMetrics.HealthMetric.CACHING_VALIDATED_WORKER_GAUGE.getMetricName()); final Gauge fileValidatedWorkerGauge = metrics.getGauges().get(BookKeeperMetrics.HealthMetric.FILE_VALIDATED_WORKER_GAUGE.getMetricName()); coordinatorBookKeeper.handleHeartbeat(TEST_HOSTNAME_WORKER1, TEST_STATUS_ALL_VALIDATED); coordinatorBookKeeper.handleHeartbeat(TEST_HOSTNAME_WORKER2, TEST_STATUS_ALL_VALIDATED); long workerCount = (long) liveWorkerGauge.getValue(); long cachingValidationCount = (long) cachingValidatedWorkerGauge.getValue(); long fileValidationCount = (long) fileValidatedWorkerGauge.getValue(); assertEquals(workerCount, 2, "Incorrect number of workers reporting heartbeat"); assertEquals(cachingValidationCount, 2, "Incorrect number of workers have been validated"); assertEquals(fileValidationCount, 2, "Incorrect number of workers have been validated"); ticker.advance(healthStatusExpiry, TimeUnit.MILLISECONDS); coordinatorBookKeeper.handleHeartbeat(TEST_HOSTNAME_WORKER1, TEST_STATUS_ALL_VALIDATED); workerCount = (long) liveWorkerGauge.getValue(); cachingValidationCount = (long) cachingValidatedWorkerGauge.getValue(); fileValidationCount = (long) fileValidatedWorkerGauge.getValue(); assertEquals(workerCount, 1, "Incorrect number of workers reporting heartbeat"); assertEquals(cachingValidationCount, 1, "Incorrect number of workers have been validated"); assertEquals(fileValidationCount, 1, "Incorrect number of workers have been validated"); } }
@Override public void handleHeartbeat(String workerHostname, HeartbeatStatus heartbeatStatus) { if (CacheConfig.isHeartbeatEnabled(conf) || !CacheConfig.isEmbeddedModeEnabled(conf)) { liveWorkerCache.put(workerHostname, true); log.debug("Received heartbeat from " + workerHostname); if (isValidationEnabled) { if (heartbeatStatus.cachingValidationSucceeded) { cachingValidatedWorkerCache.put(workerHostname, true); } else { log.error(String.format("Caching validation failed for worker node (hostname: %s)", workerHostname)); } if (heartbeatStatus.fileValidationSucceeded) { fileValidatedWorkerCache.put(workerHostname, true); } else { log.error(String.format("File validation failed for worker node (hostname: %s)", workerHostname)); } } } }
CoordinatorBookKeeper extends BookKeeper { @Override public void handleHeartbeat(String workerHostname, HeartbeatStatus heartbeatStatus) { if (CacheConfig.isHeartbeatEnabled(conf) || !CacheConfig.isEmbeddedModeEnabled(conf)) { liveWorkerCache.put(workerHostname, true); log.debug("Received heartbeat from " + workerHostname); if (isValidationEnabled) { if (heartbeatStatus.cachingValidationSucceeded) { cachingValidatedWorkerCache.put(workerHostname, true); } else { log.error(String.format("Caching validation failed for worker node (hostname: %s)", workerHostname)); } if (heartbeatStatus.fileValidationSucceeded) { fileValidatedWorkerCache.put(workerHostname, true); } else { log.error(String.format("File validation failed for worker node (hostname: %s)", workerHostname)); } } } } }
CoordinatorBookKeeper extends BookKeeper { @Override public void handleHeartbeat(String workerHostname, HeartbeatStatus heartbeatStatus) { if (CacheConfig.isHeartbeatEnabled(conf) || !CacheConfig.isEmbeddedModeEnabled(conf)) { liveWorkerCache.put(workerHostname, true); log.debug("Received heartbeat from " + workerHostname); if (isValidationEnabled) { if (heartbeatStatus.cachingValidationSucceeded) { cachingValidatedWorkerCache.put(workerHostname, true); } else { log.error(String.format("Caching validation failed for worker node (hostname: %s)", workerHostname)); } if (heartbeatStatus.fileValidationSucceeded) { fileValidatedWorkerCache.put(workerHostname, true); } else { log.error(String.format("File validation failed for worker node (hostname: %s)", workerHostname)); } } } } CoordinatorBookKeeper(Configuration conf, BookKeeperMetrics bookKeeperMetrics); @VisibleForTesting  CoordinatorBookKeeper(Configuration conf, BookKeeperMetrics bookKeeperMetrics, Ticker ticker); }
CoordinatorBookKeeper extends BookKeeper { @Override public void handleHeartbeat(String workerHostname, HeartbeatStatus heartbeatStatus) { if (CacheConfig.isHeartbeatEnabled(conf) || !CacheConfig.isEmbeddedModeEnabled(conf)) { liveWorkerCache.put(workerHostname, true); log.debug("Received heartbeat from " + workerHostname); if (isValidationEnabled) { if (heartbeatStatus.cachingValidationSucceeded) { cachingValidatedWorkerCache.put(workerHostname, true); } else { log.error(String.format("Caching validation failed for worker node (hostname: %s)", workerHostname)); } if (heartbeatStatus.fileValidationSucceeded) { fileValidatedWorkerCache.put(workerHostname, true); } else { log.error(String.format("File validation failed for worker node (hostname: %s)", workerHostname)); } } } } CoordinatorBookKeeper(Configuration conf, BookKeeperMetrics bookKeeperMetrics); @VisibleForTesting  CoordinatorBookKeeper(Configuration conf, BookKeeperMetrics bookKeeperMetrics, Ticker ticker); @Override void handleHeartbeat(String workerHostname, HeartbeatStatus heartbeatStatus); }
CoordinatorBookKeeper extends BookKeeper { @Override public void handleHeartbeat(String workerHostname, HeartbeatStatus heartbeatStatus) { if (CacheConfig.isHeartbeatEnabled(conf) || !CacheConfig.isEmbeddedModeEnabled(conf)) { liveWorkerCache.put(workerHostname, true); log.debug("Received heartbeat from " + workerHostname); if (isValidationEnabled) { if (heartbeatStatus.cachingValidationSucceeded) { cachingValidatedWorkerCache.put(workerHostname, true); } else { log.error(String.format("Caching validation failed for worker node (hostname: %s)", workerHostname)); } if (heartbeatStatus.fileValidationSucceeded) { fileValidatedWorkerCache.put(workerHostname, true); } else { log.error(String.format("File validation failed for worker node (hostname: %s)", workerHostname)); } } } } CoordinatorBookKeeper(Configuration conf, BookKeeperMetrics bookKeeperMetrics); @VisibleForTesting  CoordinatorBookKeeper(Configuration conf, BookKeeperMetrics bookKeeperMetrics, Ticker ticker); @Override void handleHeartbeat(String workerHostname, HeartbeatStatus heartbeatStatus);  }