Stream Enrich: use Kafka callback based API to detect failures to send messages

[simplesteph]

KafkaProducer.send is asynchronous and I don't know if that will fail directly. It seems you need to provide a callback to catch exceptions:

https://kafka.apache.org/0100/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html
(send method)

Fully non-blocking usage can make use of the Callback parameter to provide a callback that will be invoked when the request is complete.

 ProducerRecord<byte[],byte[]> record = new ProducerRecord<byte[],byte[]>("the-topic", key, value);
 producer.send(myRecord,
               new Callback() {
                   public void onCompletion(RecordMetadata metadata, Exception e) {
                       if(e != null)
                           e.printStackTrace();
                       System.out.println("The offset of the record we just sent is: " + metadata.offset());
                   }
               });

vs

snowplow/2-collectors/scala-stream-collector/src/main/scala/com.snowplowanalytics.snowplow.collectors/scalastream/sinks/KafkaSink.scala

Line 82 in e977c97

try {

[alexanderdean]

Good catch!