package com.ctrip.hermes.kafka.util;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.metrics.KafkaMetric;
import org.apache.kafka.common.metrics.MetricsReporter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ctrip.hermes.core.utils.HermesThreadFactory;

public class DebugLoggerMetricsReporter implements MetricsReporter {

	private static final Logger m_logger = LoggerFactory.getLogger(DebugLoggerMetricsReporter.class);

	private Map<MetricName, KafkaMetric> metrics = new HashMap<MetricName, KafkaMetric>();

	private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(HermesThreadFactory
	      .create("MetricsLogger", true));

	@Override
	public void configure(Map<String, ?> configs) {
		long millis = TimeUnit.MINUTES.toMillis(1);
		scheduler.scheduleAtFixedRate(new Runnable() {

			@Override
			public void run() {
				for (Map.Entry<MetricName, KafkaMetric> e : metrics.entrySet()) {
					m_logger.debug("{} : {}", e.getKey().name(), e.getValue().value());
				}
			}
		}, millis, millis, TimeUnit.MILLISECONDS);
	}

	private void addMetric(KafkaMetric metric) {
		metrics.put(metric.metricName(), metric);
	}

	@Override
	public void init(List<KafkaMetric> metrics) {
		for (KafkaMetric metric : metrics) {
			addMetric(metric);
		}
	}

	@Override
	public void metricChange(KafkaMetric metric) {
		addMetric(metric);
	}

	@Override
	public void close() {
		metrics.clear();
	}

}

@@ -768,22 +768,16 @@
 private 
-final 
 Schedule
@@ -806,174 +806,385 @@
 uler
- = Executors.newSingleThreadScheduledExecutor(HermesThreadFactory%0A%09      .create(%22MetricsLogger%22, true));%0A%0A%09@Override%0A%09public void configure(Map%3CString, ?%3E configs) %7B
+;%0A%0A%09private void addMetric(KafkaMetric metric) %7B%0A%09%09metrics.put(metric.metricName(), metric);%0A%09%7D%0A%0A%09@Override%0A%09public void close() %7B%0A%09%09metrics.clear();%0A%09%09if (scheduler != null)%0A%09%09%09scheduler.shutdown();%0A%09%7D%0A%0A%09@Override%0A%09public void configure(Map%3CString, ?%3E configs) %7B%0A%09%09scheduler = Executors.newSingleThreadScheduledExecutor(HermesThreadFactory.create(%22KafkaMetricsLogger%22, true));
 %0A%09%09l
@@ -1519,102 +1519,8 @@
 %09%7D%0A%0A
-%09private void addMetric(KafkaMetric metric) %7B%0A%09%09metrics.put(metric.metricName(), metric);%0A%09%7D%0A%0A
 %09@Ov
@@ -1730,63 +1730,6 @@
 %09%7D%0A%0A
-%09@Override%0A%09public void close() %7B%0A%09%09metrics.clear();%0A%09%7D%0A%0A
 %7D%0A

Enable DebugLoggerMetricReporter as configured