@SuppressWarnings("NPathComplexity")
    @Override
    public void run() {
        LOGGER.info("Starting auto balancer metrics reporter with reporting interval of {} ms.", reportingIntervalMs);

        try {
            while (!shutdown) {
                long now = System.currentTimeMillis();
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("Reporting metrics for time {}.", now);
                }
                try {
                    if (now > lastReportingTime + reportingIntervalMs) {
                        numMetricSendFailure = 0;
                        lastReportingTime = now;
                        reportMetrics(now);
                    }
                    try {
                        producer.flush();
                    } catch (InterruptException ie) {
                        if (shutdown) {
                            LOGGER.info("auto balancer metric reporter is interrupted during flush due to shutdown request.");
                        } else {
                            throw ie;
                        }
                    }
                } catch (Exception e) {
                    LOGGER.error("Got exception in auto balancer metrics reporter", e);
                }
                // Log failures if there is any.
                if (numMetricSendFailure > 0) {
                    LOGGER.warn("Failed to send {} metrics for time {}", numMetricSendFailure, now);
                }
                numMetricSendFailure = 0;
                long nextReportTime = now + reportingIntervalMs;
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("Reporting finished for time {} in {} ms. Next reporting time {}",
                            now, System.currentTimeMillis() - now, nextReportTime);
                }
                while (!shutdown && now < nextReportTime) {
                    try {
                        Thread.sleep(nextReportTime - now);
                    } catch (InterruptedException ie) {
                        // let it go
                    }
                    now = System.currentTimeMillis();
                }
            }
        } finally {
            LOGGER.info("auto balancer metrics reporter exited.");
        }
    }
On kafka metric removed

@param metric {@link KafkaMetric}
core/src/main/java/kafka/autobalancer/metricsreporter/AutoBalancerMetricsReporter.java