KafkaSubmitter implements Submitter { @Override public void submit(byte[] bytes) { submit(bytes, createProducer()); } KafkaSubmitter(ZkClientProvider zkClientProvider, String topic); @Override void submit(byte[] bytes);  }
@Test public void testSubmitIgnoresNullInput() { String anyTopic = "valueNotRelevant"; KafkaSubmitter k = new KafkaSubmitter(zkClientProvider, anyTopic); Producer<byte[], byte[]> producer = mock(Producer.class); byte[] nullData = null; k.submit(nullData, producer); verifyZeroInteractions(producer); }
@Test public void testSubmitIgnoresEmptyInput() { String anyTopic = "valueNotRelevant"; KafkaSubmitter k = new KafkaSubmitter(zkClientProvider, anyTopic); Producer<byte[], byte[]> producer = mock(Producer.class); byte[] emptyData = new byte[0]; k.submit(emptyData, producer); verifyZeroInteractions(producer); }
@Test public void testSubmit() { String anyTopic = "valueNotRelevant"; KafkaSubmitter k = new KafkaSubmitter(zkClientProvider, anyTopic); Producer<byte[], byte[]> producer = mock(Producer.class); byte[] anyData = "anyData".getBytes(); k.submit(anyData, producer); verify(producer).send(any(ProducerRecord.class)); }