class KafkaDetectorManager implements Runnable {

    @Override
    public void run() {
        log.info("Starting KafkaDetectorManager");
        metricConsumer.subscribe(Collections.singletonList(metricTopic));
        boolean continueProcessing = true;

        // See Kafka: The Definitive Guide, pp. 86 ff.
        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);
            }
        }
    }

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

}