Re: custom kafka reporters without JMX

[Alex Gray]

With the help of my colleague, we got a little further.
I created a jar from the following source code (see below), but I get a 
NoClassDefFoundError when I try to start kafka.
I somehow need to tell Kafka to use my custom reporter.
In my server.properties, I'm trying to tell it to use my custom 
reporter:
kafka.metrics.reporters=kafka.metrics.KafkaGraphiteReporter

And the jar file lives in /libs, so I know it's in the class path when 
kafka starts (maybe it's not in the correct order in the class path?).

Here is the error from kafka when I try to start it:
java.lang.NoClassDefFoundError: 
com/yammer/metrics/reporting/GraphiteReporter
	at 
kafka.metrics.KafkaGraphiteReporter.init(KafkaGraphiteReporter.java:15)
	at 
kafka.metrics.KafkaMetricsReporter$$anonfun$startReporters$1.apply(KafkaMetricsReporter.scala:60)
	at 
kafka.metrics.KafkaMetricsReporter$$anonfun$startReporters$1.apply(KafkaMetricsReporter.scala:58)
	at 
scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:34)
        at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:32)
	at 
kafka.metrics.KafkaMetricsReporter$.startReporters(KafkaMetricsReporter.scala:58)
        at kafka.Kafka$.main(Kafka.scala:36)
        at kafka.Kafka.main(Kafka.scala)

Here is the source code.  It's not really doing anything at the 
moment... I just want kafka to use it when it starts up...
=====================
package kafka.metrics;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import kafka.utils.VerifiableProperties;

import com.yammer.metrics.Metrics;
import com.yammer.metrics.core.Metric;
import com.yammer.metrics.core.MetricName;
import com.yammer.metrics.reporting.GraphiteReporter;

public class KafkaGraphiteReporter implements KafkaMetricsReporter, 
Runnable {

        public void init(VerifiableProperties arg0) {
                GraphiteReporter.enable(1, TimeUnit.MINUTES,
                                "my.graphite.server.com",
                                2003);
                System.out.println("Graphite init...");
                Thread t = new Thread(this);
                t.setDaemon(true);
                t.setName("graphitebg");
                t.start();
        }

        public void run() {
                try {
                        while(true) {
                                try {
                                        Thread.sleep(3000);
                                } catch (InterruptedException e) {              
                                        e.printStackTrace();
                                }
                                System.out.println("Graphite background thread 
executing");
                                GraphiteReporter.enable(1, TimeUnit.MINUTES,
                                                "my.graphite.server.com",
                                                2003);
                                
				Map<MetricName, Metric> allMetrics = 
Metrics.defaultRegistry().allMetrics();
                                for (MetricName key : allMetrics.keySet()) {
                                        System.out.println(key + "=" + 
allMetrics.get(key).toString());
                                }
                        }
                } catch (Exception e) {
                        e.printStackTrace();
                }
        }
}

=====================

if you have any thoughts on getting this to work, that would be awesome.

Thanks,

Alex