KafkaDetectorManager implements Runnable { @Override public void run() { log.info("Starting KafkaDetectorManager"); metricConsumer.subscribe(Collections.singletonList(metricTopic)); boolean continueProcessing = true; while (continueProcessing) { try { processMetrics(); } catch (WakeupException e) { log.info("Stopping KafkaDetectorManager"); metricConsumer.close(); anomalyProducer.flush(); anomalyProducer.close(); continueProcessing = false; } catch (Exception e) { log.error("Error processing records", e); } } } KafkaDetectorManager(
            DetectorManager detectorManager,
            Consumer<String, MappedMetricData> metricConsumer,
            Producer<String, MappedMetricData> anomalyProducer,
            String metricTopic,
            String outlierTopic,
            String breakoutTopic,
            String tracingEnabled); @Generated static void main(String[] args); @Override void run();  }
@Test public void testRun() throws Exception { val outlierMmdJson = objectMapper.writeValueAsString(outlierMMD); val breakoutMmdJson = objectMapper.writeValueAsString(breakoutMMD); for (int i = 0; i < NUM_OUTLIER_METRICS; i++) { log.info("Writing mapped metric data: {}", outlierMmdJson); kafka.helper().produceStrings(METRIC_TOPIC, outlierMmdJson); } for (int i = 0; i < NUM_BREAKOUT_METRICS; i++) { log.info("Writing mapped metric data: {}", breakoutMmdJson); kafka.helper().produceStrings(METRIC_TOPIC, breakoutMmdJson); } val managerThread = new Thread(managerUnderTest); managerThread.start(); managerThread.join(THREAD_JOIN_MILLIS); managerUnderTest.getMetricConsumer().wakeup(); val outliers = kafka.helper().consumeStrings(OUTLIER_TOPIC, NUM_OUTLIER_METRICS).get(); assertEquals(NUM_OUTLIER_METRICS, outliers.size()); val breakouts = kafka.helper().consumeStrings(BREAKOUT_TOPIC, NUM_BREAKOUT_METRICS).get(); assertEquals(NUM_BREAKOUT_METRICS, breakouts.size()); for (val outlier : outliers) { log.info("outlierDetectorResult={}", outlier); } }