KafkaReporter extends AbstractPollingReporter implements MetricProcessor<JSONObject> { @Override public void shutdown() { try { super.shutdown(); } finally { this.producer.close(); } } KafkaReporter(MetricsRegistry metricsRegistry, Properties producerProperties, String topic); void run(); @Override void processGauge(MetricName name, Gauge<?> gauge, JSONObject context); @Override void processCounter(MetricName name, Counter counter, JSONObject context); @Override void processMeter(MetricName name, Metered meter, JSONObject context); @Override void processHistogram(MetricName name, Histogram histogram, JSONObject context); @Override void processTimer(MetricName name, Timer timer, JSONObject context); @Override void shutdown();  }
@Test public void testTopicReporter() { MetricsRegistry registry = new MetricsRegistry(); Counter counter = registry.newCounter(KafkaReporterTest.class, "test-counter"); counter.inc(); Properties producerProps = new Properties(); producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConnect); producerProps.put("schema.registry.url", schemaRegistry); KafkaReporter reporter = new KafkaReporter(registry, producerProps, topic); reporter.start(1, TimeUnit.SECONDS); Properties props = new Properties(); props.put("zookeeper.connect", zkConnect); props.put("group.id", UUID.randomUUID().toString()); props.put("auto.offset.reset", "smallest"); props.put("zookeeper.session.timeout.ms", "30000"); props.put("consumer.timeout.ms", "30000"); props.put("schema.registry.url", schemaRegistry); ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(props)); KafkaStream<String, Object> messageStream = consumer.createMessageStreamsByFilter(new Whitelist(topic), 1, new StringDecoder(null), new KafkaAvroDecoder(new VerifiableProperties(props))).get(0); GenericRecord message = (GenericRecord) messageStream.iterator().next().message(); assertNotNull(message); reporter.shutdown(); }