Handling Errors and Exceptions

producer.send(record, (md, ex) -> {
  if (ex != null) handle(ex);
});

if (ex instanceof TimeoutException) {
  metrics.timeout(); retryLater(record);
}

try { producer.send(record); }
catch (SerializationException e) { dlq.send(record); }

catch (TopicAuthorizationException e) {
  log.error("Missing ACL for {}", e.unauthorizedTopics());
}

props.put("enable.idempotence", true);
props.put("retries", Integer.MAX_VALUE);

if (ex instanceof RetriableException) retry(r);
else { dlq.send(r); alert(ex); }

try { while (running) consumer.poll(d); }
catch (WakeupException e) { /* shutdown */ }
finally { consumer.close(); }

props.put("auto.offset.reset", "earliest");

new ConsumerRebalanceListener() {
  public void onPartitionsRevoked(Collection<TopicPartition> p){
    consumer.commitSync();
  }
};

streams.setUncaughtExceptionHandler(e -> {
  log.error("Fatal", e);
  return StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
});