protected void processYammerMetrics(YammerMetricProcessor.Context context) throws Exception {
        for (Map.Entry<MetricName, Metric> entry : interestedMetrics.entrySet()) {
            if (LOGGER.isTraceEnabled()) {
                LOGGER.trace("Processing yammer metric {}, scope = {}", entry.getKey(), entry.getKey().getScope());
            }
            entry.getValue().processWith(yammerMetricProcessor, entry.getKey(), context);
        }
        Iterator<Map.Entry<String, AutoBalancerMetrics>> iterator = context.getMetricMap().entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, AutoBalancerMetrics> entry = iterator.next();
            AutoBalancerMetrics metrics = entry.getValue();
            if (metrics.metricType() == MetricTypes.TOPIC_PARTITION_METRIC
                    && !metrics.getMetricValueMap().containsKey(RawMetricTypes.PARTITION_SIZE)) {
                // remove metrics for closed partition
                iterator.remove();
            }
        }
    }
Send a AutoBalancerMetric to the Kafka topic.

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