I figured out a solution to my problem with the help of a friend and thought I'd share. Pretty much what I did was implement Callback() with an overloaded constructor that accepts the my message key and value so that I can choose to do something with it should I need to. A note for anyone reading this though is that the API it states that "this callback will generally execute in the background I/O thread so it should be fast", which means it's advisable to avoid significant processing in the Callback(). The code snippet example follows: ProducerRecord<String, byte[]> message = null;
protected KafkaProducer<String, byte[]> aProducer = null;
aProducer = createKafkaProducer();
[...]
message = new ProducerRecord<String, byte[]>(producerConfig.getKafkaTopic(), byteString);
send();
[...]
public void send() {
aProducer.send(message, new MyCallback(message.key(),new String(message.value())));
}
}
//Implemented Callback but I overloaded the constructor to pass in the message so that, if needed, I can do something with the message itself on error. Since according to the API, "This callback will generally execute in the background I/O thread so it should be fast.", I should avoid expensive actions in this callback, so I would have to spin of another executor if I want to do much.
class MyCallback implements Callback {
private final String messageKey;
private final String messageValue;
public MyCallback(String messageKey, String messageValue) {
this.messageKey = messageKey;
this.messageValue = messageValue;
}
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
// This is where I will handle exceptions and I'll have the message should I need to write it out somewhere to deal with later.
[...]
}
... View more
