title: Quarkus + Kafka + Smallrye exception handling
body: <p>How can i handle exception on streaming processing with quarkus + kafka + smallrye?</p>
  
  <p>My code is very similar to the imperative producer example on quarkus guide (<a href="https://quarkus.io/guides/kafka#imperative-usage" rel="nofollow noreferrer">https://quarkus.io/guides/kafka#imperative-usage</a>)</p>
  
  <pre><code>import io.smallrye.reactive.messaging.annotations.Channel;
  import io.smallrye.reactive.messaging.annotations.Emitter;
  
  import javax.inject.Inject;
  import javax.ws.rs.POST;
  import javax.ws.rs.Path;
  import javax.ws.rs.Consumes;
  import javax.ws.rs.core.MediaType;
  
  @Path("/prices")
  public class PriceResource {
  
      @Inject @Channel("price-create") Emitter&lt;Double&gt; priceEmitter;
  
      @POST
      @Consumes(MediaType.TEXT_PLAIN)
      public void addPrice(Double price) {
          priceEmitter.send(price);
      }
  }
  </code></pre>
  
  <p>I wanted something similar the vanilla Kafka library, that gives the option to handle the callback of each record requested to send.</p>
  
  <pre><code>ProducerRecord&lt;String, String&gt; record = new ProducerRecord&lt;&gt;("topic-name", key, value);
  
  producer.send(record, new Callback() {
      @Override
      public void onCompletion(RecordMetadata metadata, Exception exception) {
          logger.info(record.toString());
  
          if (exception != null) {
              logger.error("Producer exception", exception);
          }
      }
  });
  </code></pre>
  
  <p>Tks</p>