metrics.reporters

package kafkaReporter;

import java.io.PrintWriter;
import java.util.List;
import java.util.Map;

import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.metrics.KafkaMetric;
import org.apache.kafka.common.metrics.MetricsReporter;


public class MetricsExtractor  implements MetricsReporter{
    private PrintWriter writer;

    public void configure(Map<String, ?> arg0) {

        System.out.println("Metrics Reporter Config...");
    }

    public void close() {
        writer.close();
        System.out.println("Metrics Reporter Closing...");
    }

    public void init(List<KafkaMetric> arg0) {
        System.out.println("Metrics Reporter init...");

        try {
             writer = new PrintWriter("KmetricsReporter.txt", "UTF-8");
             //writer.println("Processing a metric change for {}", kafkaMetric.metricName().toString());
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public void metricChange(KafkaMetric kafkaMetric) {
//      writer.print(kafkaMetric.metricName().toString());
//      writer.println(": " + kafkaMetric.toString());
        try {
            if(kafkaMetric.value() != 0)
            {
                writer.print(sanitizeName(kafkaMetric.metricName()));
                writer.println(": " + kafkaMetric.value());
            }
        } catch (Exception e) {
            e.printStackTrace(writer);
        }
    }

    public void metricRemoval(KafkaMetric arg0) {
        // TODO Auto-generated method stub

    }

    private String sanitizeName(MetricName name) {
        StringBuilder result = new StringBuilder().append(name.group()).append('.');
        for (Map.Entry<String, String> tag : name.tags().entrySet()) {
            result.append(tag.getValue()).append('.');
        }
        return result.append(name.name()).toString().replace(' ', '_').replace("\\.", "_");
    }
}