@Test public void testShutdown() throws InterruptedException { PriorityBlockingQueue<Anomaly> anomalies = new PriorityBlockingQueue<>(ANOMALY_DETECTOR_INITIAL_QUEUE_SIZE, anomalyComparator()); AnomalyNotifier mockAnomalyNotifier = EasyMock.createNiceMock(AnomalyNotifier.class); AdminClient mockAdminClient = EasyMock.createNiceMock(AdminClient.class); BrokerFailureDetector mockBrokerFailureDetector = EasyMock.createNiceMock(BrokerFailureDetector.class); GoalViolationDetector mockGoalViolationDetector = EasyMock.createNiceMock(GoalViolationDetector.class); MetricAnomalyDetector mockMetricAnomalyDetector = EasyMock.createNiceMock(MetricAnomalyDetector.class); TopicAnomalyDetector mockTopicAnomalyDetector = EasyMock.createNiceMock(TopicAnomalyDetector.class); MaintenanceEventDetector mockMaintenanceEventDetector = EasyMock.createNiceMock(MaintenanceEventDetector.class); DiskFailureDetector mockDiskFailureDetector = EasyMock.createNiceMock(DiskFailureDetector.class); KafkaCruiseControl mockKafkaCruiseControl = EasyMock.createNiceMock(KafkaCruiseControl.class); ScheduledExecutorService detectorScheduler = Executors.newScheduledThreadPool(2, new KafkaCruiseControlThreadFactory("AnomalyDetector", false, null)); AnomalyDetectorManager anomalyDetectorManager = new AnomalyDetectorManager(anomalies, mockAdminClient, MOCK_ANOMALY_DETECTION_INTERVAL_MS, mockKafkaCruiseControl, mockAnomalyNotifier, mockGoalViolationDetector, mockBrokerFailureDetector, mockMetricAnomalyDetector, mockDiskFailureDetector, mockTopicAnomalyDetector, mockMaintenanceEventDetector, detectorScheduler); anomalyDetectorManager.shutdown(); Thread t = new Thread(anomalyDetectorManager::shutdown); t.start(); t.join(30000L); assertEquals(0, anomalyDetectorManager.numSelfHealingStarted()); assertEquals(0, anomalyDetectorManager.numCheckedWithDelay()); assertTrue(detectorScheduler.isTerminated()); }
public void shutdown() { LOG.info("Shutting down anomaly detector."); synchronized (_shutdownLock) { _shutdown = true; } _anomalies.add(SHUTDOWN_ANOMALY); _maintenanceEventDetector.shutdown(); _detectorScheduler.shutdown(); KafkaCruiseControlUtils.closeAdminClientWithTimeout(_adminClient); try { _detectorScheduler.awaitTermination(SCHEDULER_SHUTDOWN_TIMEOUT_MS, TimeUnit.MILLISECONDS); if (!_detectorScheduler.isTerminated()) { LOG.warn("The sampling scheduler failed to shutdown in " + SCHEDULER_SHUTDOWN_TIMEOUT_MS + " ms."); } } catch (InterruptedException e) { LOG.warn("Interrupted while waiting for anomaly detector to shutdown."); } _brokerFailureDetector.shutdown(); _anomalyLoggerExecutor.shutdownNow(); LOG.info("Anomaly detector shutdown completed."); }
AnomalyDetectorManager { public void shutdown() { LOG.info("Shutting down anomaly detector."); synchronized (_shutdownLock) { _shutdown = true; } _anomalies.add(SHUTDOWN_ANOMALY); _maintenanceEventDetector.shutdown(); _detectorScheduler.shutdown(); KafkaCruiseControlUtils.closeAdminClientWithTimeout(_adminClient); try { _detectorScheduler.awaitTermination(SCHEDULER_SHUTDOWN_TIMEOUT_MS, TimeUnit.MILLISECONDS); if (!_detectorScheduler.isTerminated()) { LOG.warn("The sampling scheduler failed to shutdown in " + SCHEDULER_SHUTDOWN_TIMEOUT_MS + " ms."); } } catch (InterruptedException e) { LOG.warn("Interrupted while waiting for anomaly detector to shutdown."); } _brokerFailureDetector.shutdown(); _anomalyLoggerExecutor.shutdownNow(); LOG.info("Anomaly detector shutdown completed."); } }
AnomalyDetectorManager { public void shutdown() { LOG.info("Shutting down anomaly detector."); synchronized (_shutdownLock) { _shutdown = true; } _anomalies.add(SHUTDOWN_ANOMALY); _maintenanceEventDetector.shutdown(); _detectorScheduler.shutdown(); KafkaCruiseControlUtils.closeAdminClientWithTimeout(_adminClient); try { _detectorScheduler.awaitTermination(SCHEDULER_SHUTDOWN_TIMEOUT_MS, TimeUnit.MILLISECONDS); if (!_detectorScheduler.isTerminated()) { LOG.warn("The sampling scheduler failed to shutdown in " + SCHEDULER_SHUTDOWN_TIMEOUT_MS + " ms."); } } catch (InterruptedException e) { LOG.warn("Interrupted while waiting for anomaly detector to shutdown."); } _brokerFailureDetector.shutdown(); _anomalyLoggerExecutor.shutdownNow(); LOG.info("Anomaly detector shutdown completed."); } AnomalyDetectorManager(KafkaCruiseControl kafkaCruiseControl, Time time, MetricRegistry dropwizardMetricRegistry);   AnomalyDetectorManager(PriorityBlockingQueue<Anomaly> anomalies,
                         AdminClient adminClient,
                         long anomalyDetectionIntervalMs,
                         KafkaCruiseControl kafkaCruiseControl,
                         AnomalyNotifier anomalyNotifier,
                         GoalViolationDetector goalViolationDetector,
                         BrokerFailureDetector brokerFailureDetector,
                         MetricAnomalyDetector metricAnomalyDetector,
                         DiskFailureDetector diskFailureDetector,
                         TopicAnomalyDetector topicAnomalyDetector,
                         MaintenanceEventDetector maintenanceEventDetector,
                         ScheduledExecutorService detectorScheduler); }
AnomalyDetectorManager { public void shutdown() { LOG.info("Shutting down anomaly detector."); synchronized (_shutdownLock) { _shutdown = true; } _anomalies.add(SHUTDOWN_ANOMALY); _maintenanceEventDetector.shutdown(); _detectorScheduler.shutdown(); KafkaCruiseControlUtils.closeAdminClientWithTimeout(_adminClient); try { _detectorScheduler.awaitTermination(SCHEDULER_SHUTDOWN_TIMEOUT_MS, TimeUnit.MILLISECONDS); if (!_detectorScheduler.isTerminated()) { LOG.warn("The sampling scheduler failed to shutdown in " + SCHEDULER_SHUTDOWN_TIMEOUT_MS + " ms."); } } catch (InterruptedException e) { LOG.warn("Interrupted while waiting for anomaly detector to shutdown."); } _brokerFailureDetector.shutdown(); _anomalyLoggerExecutor.shutdownNow(); LOG.info("Anomaly detector shutdown completed."); } AnomalyDetectorManager(KafkaCruiseControl kafkaCruiseControl, Time time, MetricRegistry dropwizardMetricRegistry);   AnomalyDetectorManager(PriorityBlockingQueue<Anomaly> anomalies,
                         AdminClient adminClient,
                         long anomalyDetectionIntervalMs,
                         KafkaCruiseControl kafkaCruiseControl,
                         AnomalyNotifier anomalyNotifier,
                         GoalViolationDetector goalViolationDetector,
                         BrokerFailureDetector brokerFailureDetector,
                         MetricAnomalyDetector metricAnomalyDetector,
                         DiskFailureDetector diskFailureDetector,
                         TopicAnomalyDetector topicAnomalyDetector,
                         MaintenanceEventDetector maintenanceEventDetector,
                         ScheduledExecutorService detectorScheduler); void startDetection(); void shutdown(); synchronized AnomalyDetectorState anomalyDetectorState(); void maybeClearOngoingAnomalyDetectionTimeMs(); void resetHasUnfixableGoals(); boolean setSelfHealingFor(AnomalyType anomalyType, boolean isSelfHealingEnabled); long numCheckedWithDelay(); void markSelfHealingFinished(String anomalyId); }
AnomalyDetectorManager { public void shutdown() { LOG.info("Shutting down anomaly detector."); synchronized (_shutdownLock) { _shutdown = true; } _anomalies.add(SHUTDOWN_ANOMALY); _maintenanceEventDetector.shutdown(); _detectorScheduler.shutdown(); KafkaCruiseControlUtils.closeAdminClientWithTimeout(_adminClient); try { _detectorScheduler.awaitTermination(SCHEDULER_SHUTDOWN_TIMEOUT_MS, TimeUnit.MILLISECONDS); if (!_detectorScheduler.isTerminated()) { LOG.warn("The sampling scheduler failed to shutdown in " + SCHEDULER_SHUTDOWN_TIMEOUT_MS + " ms."); } } catch (InterruptedException e) { LOG.warn("Interrupted while waiting for anomaly detector to shutdown."); } _brokerFailureDetector.shutdown(); _anomalyLoggerExecutor.shutdownNow(); LOG.info("Anomaly detector shutdown completed."); } AnomalyDetectorManager(KafkaCruiseControl kafkaCruiseControl, Time time, MetricRegistry dropwizardMetricRegistry);   AnomalyDetectorManager(PriorityBlockingQueue<Anomaly> anomalies,
                         AdminClient adminClient,
                         long anomalyDetectionIntervalMs,
                         KafkaCruiseControl kafkaCruiseControl,
                         AnomalyNotifier anomalyNotifier,
                         GoalViolationDetector goalViolationDetector,
                         BrokerFailureDetector brokerFailureDetector,
                         MetricAnomalyDetector metricAnomalyDetector,
                         DiskFailureDetector diskFailureDetector,
                         TopicAnomalyDetector topicAnomalyDetector,
                         MaintenanceEventDetector maintenanceEventDetector,
                         ScheduledExecutorService detectorScheduler); void startDetection(); void shutdown(); synchronized AnomalyDetectorState anomalyDetectorState(); void maybeClearOngoingAnomalyDetectionTimeMs(); void resetHasUnfixableGoals(); boolean setSelfHealingFor(AnomalyType anomalyType, boolean isSelfHealingEnabled); long numCheckedWithDelay(); void markSelfHealingFinished(String anomalyId);  }