@Test public void testFilterResultSerializer() throws Exception { KafkaWriter2 writer = new KafkaWriter2( producerMock, TOPIC, mock(ResultSerializer.class)); writer.doWrite(dummyServer(), dummyQuery(), dummyResults()); verifyNoMoreInteractions(producerMock); }
@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();  }