class 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);
    }

    public  KafkaDetectorManager(
            DetectorManager detectorManager,
            Consumer<String, MappedMetricData> metricConsumer,
            Producer<String, MappedMetricData> anomalyProducer,
            String metricTopic,
            String outlierTopic,
            String breakoutTopic,
            String tracingEnabled);

}