class AnomalyToMetricMapper {

    public MetricData toMetricData(MappedMetricData anomaly) {
        if (anomaly == null) {
            return null;
        }

        val metricData = anomaly.getMetricData();
        val metricDef = metricData.getMetricDefinition();
        val tags = metricDef.getTags();
        val kvTags = tags.getKv();

        // Reverting back to returning null instead of generating an exception. We get to define the transform we want
        // here, and I don't see an advantage to generating an exception when the proper transform at the Kafka level
        // is to transform it into a null too. It reduces the burden on the client. [WLW]
        if (kvTags.containsKey(AA_DETECTOR_UUID) || kvTags.containsKey(AA_ANOMALY_LEVEL)) {
            return null;
        }

        val detectorUuid = anomaly.getDetectorUuid();
        val outlierResult = (OutlierDetectorResult) anomaly.getAnomalyResult();
        val outlierLevel = outlierResult.getAnomalyLevel();

        AssertUtil.notNull(detectorUuid, "detectorUuid can't be null");
        AssertUtil.notNull(outlierLevel, "anomalyLevel can't be null");

        val newKVTags = new HashMap<>(kvTags);
        newKVTags.put(AA_DETECTOR_UUID, detectorUuid.toString());
        newKVTags.put(AA_ANOMALY_LEVEL, outlierLevel.toString());

        val newKey = metricDef.getKey();
        val newTags = new TagCollection(newKVTags, Collections.EMPTY_SET);
        val newMeta = new TagCollection(Collections.EMPTY_MAP);
        val newMetricDef = new MetricDefinition(newKey, newTags, newMeta);
        val value = metricData.getValue();
        val timestamp = metricData.getTimestamp();

        return new MetricData(newMetricDef, value, timestamp);
    }

}