class AnomaliesProcessor {

    public List<AnomalyModel> processMetrics(ConsumerRecords<String, MappedMetricData> metricRecords, ExecutorService executorService) {

        List<AnomalyModel> anomalyModels = new ArrayList();
        for (ConsumerRecord<String, MappedMetricData> consumerRecord : metricRecords) {
            MappedMetricData mappedMetricData = consumerRecord.value();
            if (mappedMetricData != null) {
                OutlierDetectorResult outlierDetectorResult = (OutlierDetectorResult) mappedMetricData.getAnomalyResult();
                if (outlierDetectorResult.getAnomalyLevel() != AnomalyLevel.NORMAL) {
                    AnomalyModel.Builder anomalyModel = AnomalyModel.newBuilder();
                    if (outlierDetectorResult != null) {
                        anomalyModel.level(outlierDetectorResult.getAnomalyLevel().toString());
                        anomalyModel.anomalyThresholds(outlierDetectorResult.getThresholds());
                    }

                    if (mappedMetricData.getDetectorUuid() != null) {
                        anomalyModel.uuid(mappedMetricData.getDetectorUuid().toString());
                    }
                    MetricData metricData = mappedMetricData.getMetricData();
                    if (metricData != null) {
                        anomalyModel.timestamp(VisualizerUtility.convertToDate(metricData.getTimestamp()));
                        anomalyModel.value(metricData.getValue());
                        if (metricData.getMetricDefinition() != null) {
                            anomalyModel.key(metricData.getMetricDefinition().getKey());
                            anomalyModel.tags(metricData.getMetricDefinition().getTags());
                        }
                    }
                    anomalyModels.add(anomalyModel.build());
                }
            }
        }
        if (anomalyModels.size() > 0) {
            ElasticSearchBulkService elasticSearchBulkService = new ElasticSearchBulkService(anomalyModels);
            executorService.submit(elasticSearchBulkService);
            log.info("sending anomaly records to elasticsearch: {}", anomalyModels.size());
        }
        return anomalyModels;
    }

    public  AnomaliesProcessor();

}