@Override
    public void processHistogram(MetricName metricName, Histogram histogram, Context context) {
        if (MetricsUtils.isInterested(metricName)) {
            if (LOG.isTraceEnabled()) {
                LOG.trace("Processing metric {} of type Histogram.", metricName);
            }
            // Get max metric value
            AutoBalancerMetrics ccm = MetricsUtils.toAutoBalancerMetric(context.time(),
                    context.brokerId(),
                    context.brokerRack(),
                    metricName,
                    histogram.max(),
                    MetricsUtils.ATTRIBUTE_MAX);
            context.merge(ccm);

            // Get mean metric value
            ccm = MetricsUtils.toAutoBalancerMetric(context.time(),
                    context.brokerId(),
                    context.brokerRack(),
                    metricName,
                    histogram.mean(),
                    MetricsUtils.ATTRIBUTE_MEAN);
            context.merge(ccm);

            Snapshot snapshot = histogram.getSnapshot();
            // Get 50th percentile (i.e. median) metric value
            ccm = MetricsUtils.toAutoBalancerMetric(context.time(),
                    context.brokerId(),
                    context.brokerRack(),
                    metricName,
                    snapshot.getMedian(),
                    MetricsUtils.ATTRIBUTE_50TH_PERCENTILE);
            context.merge(ccm);

            // Get 999th percentile metric value
            ccm = MetricsUtils.toAutoBalancerMetric(context.time(),
                    context.brokerId(),
                    context.brokerRack(),
                    metricName,
                    snapshot.get999thPercentile(),
                    MetricsUtils.ATTRIBUTE_999TH_PERCENTILE);
            context.merge(ccm);
        }
    }
This class was modified based on Cruise Control: com.linkedin.kafka.cruisecontrol.metricsreporter.metric.YammerMetricProcessor.
core/src/main/java/kafka/autobalancer/metricsreporter/metric/YammerMetricProcessor.java