KafkaReporter extends Reporter { @Override public void report(Measurement measurement) { if (producer == null) { logger.warn("Kafka producer is not initialized."); return; } producer.send(new ProducerRecord<>(topic, partitionKey, measurement.toJson())); } KafkaReporter(ReporterConfiguration configuration, GlobalConfiguration globalConfiguration); @Override void report(Measurement measurement); @Override void stop();  }
@Test public void send_measurements() { ReporterConfiguration config = new ReporterConfiguration(); config.options.put("kafkaBootstrapServers", HOST + ":" + BROKER_PORT); config.options.put("kafkaTopic", "testTopic"); KafkaReporter reporter = new KafkaReporter(config, GlobalConfiguration.getDefault()); Map<String, String> tags = new HashMap<>(); tags.put("tag1", "tv1"); tags.put("tag2", "tv2"); Map<String, String> fields = new HashMap<>(); fields.put("v2", "abc"); Measurement measurement = Measurement.createSimple("m1", 1.0, 1434055662, TimeUnit.SECONDS, tags, fields); reporter.report(measurement); Properties consumerProps = new Properties(); consumerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, HOST + ":" + BROKER_PORT); consumerProps.put("group.id", "test-group"); consumerProps.put("client.id", "test-consumer"); consumerProps.put("auto.offset.reset", "earliest"); consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps); consumer.subscribe(Collections.singletonList("testTopic")); ConsumerRecords<String, String> records = consumer.poll(30000); assertEquals(1, records.count()); Iterator<ConsumerRecord<String, String>> recordIterator = records.iterator(); ConsumerRecord<String, String> record = recordIterator.next(); assertEquals(Whitebox.getInternalState(reporter, "partitionKey"), record.key()); assertEquals(measurement.toJson(), record.value()); }