KafkaDetectorManager implements Runnable { static KafkaDetectorManager buildManager(Config config) { MetricRegistry metricRegistry = getMetricRegistry(); val detectorSource = DetectorUtil.buildDetectorSource(config); val dataInitializer = DataInitializerFactory.buildDataInitializer(config); val detectorManager = new DetectorManager(detectorSource, dataInitializer, config, metricRegistry); val metricConsumerConfig = config.getConfig(METRIC_CONSUMER); val metricConsumerProps = ConfigUtil.toConsumerConfig(metricConsumerConfig); val anomalyProducerConfig = config.getConfig(ANOMALY_PRODUCER); val anomalyProducerProps = ConfigUtil.toProducerConfig(anomalyProducerConfig); val tracingDetectorManagerConfig = config.getConfig(DM_TRACING); val tracingEnabled = tracingDetectorManagerConfig.getString(TRACING_STATUS_STRING); if (tracingEnabled.equals(TRACING_STATUS_CHECK_STRING)){ metricConsumerProps.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingConsumerInterceptor.class.getName()); anomalyProducerProps.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingProducerInterceptor.class.getName()); val collectorHeaders = new HashMap<String, String>(); GlobalTracer.registerIfAbsent(TracingUtil.getTracer(collectorHeaders, tracingDetectorManagerConfig)); } val metricConsumer = new KafkaConsumer<String, MappedMetricData>(metricConsumerProps); val metricConsumerTopic = metricConsumerConfig.getString(TOPIC); val anomalyProducer = new KafkaProducer<String, MappedMetricData>(anomalyProducerProps); val anomalyProducerOutlierTopic = anomalyProducerConfig.getString(OUTLIER_TOPIC); val anomalyProducerBreakoutTopic = anomalyProducerConfig.getString(BREAKOUT_TOPIC); return new KafkaDetectorManager( detectorManager, metricConsumer, anomalyProducer, metricConsumerTopic, anomalyProducerOutlierTopic, anomalyProducerBreakoutTopic, tracingEnabled); } 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 testBuildManager() { val config = ConfigFactory.load("detector-manager.conf"); val manager = KafkaDetectorManager.buildManager(config); assertNotNull(manager); assertEquals(METRIC_TOPIC, manager.getMetricTopic()); assertEquals(OUTLIER_TOPIC, manager.getOutlierTopic()); assertEquals(BREAKOUT_TOPIC, manager.getBreakoutTopic()); assertEquals(TRACING_ENABLED, manager.getTracingEnabled()); }