protected void checkMetricCompleteness(YammerMetricProcessor.Context context) {
        for (AutoBalancerMetrics metrics : context.getMetricMap().values()) {
            if (metrics.metricType() == MetricTypes.TOPIC_PARTITION_METRIC
                    && !MetricsUtils.sanityCheckTopicPartitionMetricsCompleteness(metrics)) {
                throw new IllegalStateException("Missing metrics for topic partition " + metrics.key());
            }
            if (metrics.metricType() == MetricTypes.BROKER_METRIC
                    && !MetricsUtils.sanityCheckBrokerMetricsCompleteness(metrics)) {
                throw new IllegalStateException("Missing metrics for broker " + metrics.key());
            }
        }
    }
Send a AutoBalancerMetric to the Kafka topic.

@param ccm the auto balancer metric to send.
core/src/main/java/kafka/autobalancer/metricsreporter/AutoBalancerMetricsReporter.java