@Test public void testWrite() throws Exception { KafkaWriter2 writer = new KafkaWriter2( producerMock, TOPIC, new DetailedResultSerializer()); ArgumentCaptor<ProducerRecord> recordCaptor = ArgumentCaptor.forClass(ProducerRecord.class); when(producerMock.send(recordCaptor.capture())).thenReturn(null); writer.doWrite(dummyServer(), null, dummyResults()); verify(producerMock, times(3)).send(any(ProducerRecord.class)); List<ProducerRecord> records = recordCaptor.getAllValues(); assertThat(records).hasSize(3); for(ProducerRecord<String, String> record: records) { assertThat(record.topic()).isEqualTo("topic"); JsonNode jsonNode = objectMapper.readValue(record.value(), JsonNode.class); assertThat(jsonNode.get("host").asText()).isEqualTo(DEFAULT_HOST); assertThat(jsonNode.get("epoch").asLong()).isEqualTo(0L); assertThat(jsonNode.get("value").asLong()).isNotNull(); } }
@Override public void doWrite(Server server, Query query, Iterable<Result> results) throws Exception { for (Result result : results) { String message = resultSerializer.serialize(server, query, result); if (message != null) { producer.send(new ProducerRecord<String, String>(topic, message)); } } }
KafkaWriter2 extends OutputWriterAdapter { @Override public void doWrite(Server server, Query query, Iterable<Result> results) throws Exception { for (Result result : results) { String message = resultSerializer.serialize(server, query, result); if (message != null) { producer.send(new ProducerRecord<String, String>(topic, message)); } } } }
KafkaWriter2 extends OutputWriterAdapter { @Override public void doWrite(Server server, Query query, Iterable<Result> results) throws Exception { for (Result result : results) { String message = resultSerializer.serialize(server, query, result); if (message != null) { producer.send(new ProducerRecord<String, String>(topic, message)); } } } KafkaWriter2(@Nonnull Map<String, Object> producerConfig, @Nonnull String topic, @Nonnull ResultSerializer resultSerializer); @VisibleForTesting  KafkaWriter2(Producer<String, String> producer, @Nonnull String topic, @Nonnull ResultSerializer resultSerializer); }
KafkaWriter2 extends OutputWriterAdapter { @Override public void doWrite(Server server, Query query, Iterable<Result> results) throws Exception { for (Result result : results) { String message = resultSerializer.serialize(server, query, result); if (message != null) { producer.send(new ProducerRecord<String, String>(topic, message)); } } } KafkaWriter2(@Nonnull Map<String, Object> producerConfig, @Nonnull String topic, @Nonnull ResultSerializer resultSerializer); @VisibleForTesting  KafkaWriter2(Producer<String, String> producer, @Nonnull String topic, @Nonnull ResultSerializer resultSerializer); @Override void doWrite(Server server, Query query, Iterable<Result> results); @Override void close(); }
KafkaWriter2 extends OutputWriterAdapter { @Override public void doWrite(Server server, Query query, Iterable<Result> results) throws Exception { for (Result result : results) { String message = resultSerializer.serialize(server, query, result); if (message != null) { producer.send(new ProducerRecord<String, String>(topic, message)); } } } KafkaWriter2(@Nonnull Map<String, Object> producerConfig, @Nonnull String topic, @Nonnull ResultSerializer resultSerializer); @VisibleForTesting  KafkaWriter2(Producer<String, String> producer, @Nonnull String topic, @Nonnull ResultSerializer resultSerializer); @Override void doWrite(Server server, Query query, Iterable<Result> results); @Override void close();  }