@Test public void errorReporting() throws Exception { final ExecutionException error = new ExecutionException("Emulate kafka send error", new IOException()); Future promise = mock(Future.class); expect(promise.get()).andThrow(error).anyTimes(); replay(promise); expect(kafkaProducer.send(anyObject(), anyObject(Callback.class))) .andAnswer(new IAnswer<Future<RecordMetadata>>() { @Override public Future<RecordMetadata> answer() { Callback callback = (Callback) getCurrentArguments()[1]; callback.onCompletion(null, error); return promise; } }); replay(kafkaProducer); subject.sendMessageAndTrack(TOPIC, makePayload()); verify(kafkaProducer); }
public void sendMessageAndTrack(String topic, Message message) { produce(encode(topic, message), new SendStatusCallback(this, topic, message)); }
KafkaProducerService implements IKafkaProducerService { public void sendMessageAndTrack(String topic, Message message) { produce(encode(topic, message), new SendStatusCallback(this, topic, message)); } }
KafkaProducerService implements IKafkaProducerService { public void sendMessageAndTrack(String topic, Message message) { produce(encode(topic, message), new SendStatusCallback(this, topic, message)); }  }
KafkaProducerService implements IKafkaProducerService { public void sendMessageAndTrack(String topic, Message message) { produce(encode(topic, message), new SendStatusCallback(this, topic, message)); }  @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 void sendMessageAndTrack(String topic, Message message) { produce(encode(topic, message), new SendStatusCallback(this, topic, message)); }  @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();  }