@Test public void errorDetection() throws Exception { Future promise = mock(Future.class); final ExecutionException error = new ExecutionException("Emulate kafka send error", new IOException()); expect(promise.get()).andThrow(error).anyTimes(); replay(promise); expect(kafkaProducer.send(anyObject(), anyObject(Callback.class))).andReturn(promise); replay(kafkaProducer); SendStatus status = subject.sendMessage(TOPIC, makePayload()); verify(kafkaProducer); Boolean isThrown; try { status.waitTillComplete(); isThrown = false; } catch (ExecutionException e) { isThrown = true; } Assert.assertTrue(String.format( "Exception was not thrown by %s object", status.getClass().getCanonicalName()), isThrown); }
public SendStatus sendMessage(String topic, Message message) { SendStatus sendStatus = produce(encode(topic, message), null); return sendStatus; }
KafkaProducerService implements IKafkaProducerService { public SendStatus sendMessage(String topic, Message message) { SendStatus sendStatus = produce(encode(topic, message), null); return sendStatus; } }
KafkaProducerService implements IKafkaProducerService { public SendStatus sendMessage(String topic, Message message) { SendStatus sendStatus = produce(encode(topic, message), null); return sendStatus; }  }
KafkaProducerService implements IKafkaProducerService { public SendStatus sendMessage(String topic, Message message) { SendStatus sendStatus = produce(encode(topic, message), null); return sendStatus; }  @Override void setup(FloodlightModuleContext moduleContext); void sendMessageAndTrack(String topic, Message message); void sendMessageAndTrack(String topic, String key, Message message); @Override void sendMessageAndTrack(String topic, String key, AbstractMessage message); SendStatus sendMessage(String topic, Message message); int getFailedSendMessageCounter(); }
KafkaProducerService implements IKafkaProducerService { public SendStatus sendMessage(String topic, Message message) { SendStatus sendStatus = produce(encode(topic, message), null); return sendStatus; }  @Override void setup(FloodlightModuleContext moduleContext); void sendMessageAndTrack(String topic, Message message); void sendMessageAndTrack(String topic, String key, Message message); @Override void sendMessageAndTrack(String topic, String key, AbstractMessage message); SendStatus sendMessage(String topic, Message message); int getFailedSendMessageCounter();  }