private void reportMetrics(long now) throws Exception {
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("Reporting metrics.");
        }

        YammerMetricProcessor.Context context = new YammerMetricProcessor.Context(now, brokerId, brokerRack, reportingIntervalMs);
        processMetrics(context);
        addMissingMetrics(context);
        checkMetricCompleteness(context);
        for (Map.Entry<String, AutoBalancerMetrics> entry : context.getMetricMap().entrySet()) {
            sendAutoBalancerMetric(entry.getValue());
        }

        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("Finished reporting metrics, total metrics size: {}, merged size: {}.", interestedMetrics.size(), context.getMetricMap().size());
        }
    }
Send a AutoBalancerMetric to the Kafka topic.

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