Repository: smallrye/smallrye-reactive-messaging
Issue: ClassCastException when a nacked message is sent to the DLQ

The failure strategy 'dead-letter-queue' seems to be broke.

Here is my configuration : 


Producer code : 

```
mp.messaging.outgoing.generated-price.connector=smallrye-kafka
mp.messaging.outgoing.generated-price.topic=prices
mp.messaging.outgoing.generated-price.value.serializer=org.apache.kafka.common.serialization.IntegerSerializer
```

```java
@Outgoing("generated-price")
    public Flowable<Integer> generate() {

        return Flowable.interval(5, TimeUnit.SECONDS)
                .map(tick -> random.nextInt(100));


    }
```

Consumer code : 

```
mp.messaging.incoming.prices.connector=smallrye-kafka
mp.messaging.incoming.prices.bootstrap.servers=localhost:9092
mp.messaging.incoming.prices.value.deserializer=org.apache.kafka.common.serialization.IntegerDeserializer
mp.messaging.incoming.prices.group.id=prices-group-local
mp.messaging.incoming.prices.client.id=prices-client-local
mp.messaging.incoming.prices.enable.auto.commit=false
mp.messaging.incoming.prices.failure-strategy=dead-letter-queue
```

```java
@Incoming( "prices" )
    @Outgoing( "double-prices" )
    @Acknowledgment(Acknowledgment.Strategy.POST_PROCESSING)
    public Uni< Integer > consume( final Integer price) {

        log.info("incoming message {}", price);
    
        if(price< 50) {
            return Uni.createFrom().item(price* 2);
        }
        else {
            return Uni.createFrom().failure(new Exception("Price has to be < 50"));
        }
    }
```


I got the following stacktrace when the price is >= 50 : 

```java
2020-07-21 14:19:07,770 ERROR [io.sma.rea.mes.kafka] (vert.x-eventloop-thread-8) SRMSG18207: Unable to dispatch message to Kafka: java.lang.ClassCastException: io.vertx.kafka.client.producer.RecordMetadata cannot be cast to java.lang.Void
        at io.smallrye.context.SmallRyeThreadContext$ContextualFunction.apply(SmallRyeThreadContext.java:76)
        at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:616)
        at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:591)
        at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)
        at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975)
        at io.smallrye.mutiny.operators.UniSubscribeToCompletionStage$2.onItem(UniSubscribeToCompletionStage.java:44)
        at io.smallrye.mutiny.context.ContextPropagationUniInterceptor$1.lambda$onItem$1(ContextPropagationUniInterceptor.java:35)
        at io.smallrye.context.SmallRyeThreadContext.lambda$withContext$0(SmallRyeThreadContext.java:217)
        at io.smallrye.mutiny.context.ContextPropagationUniInterceptor$1.onItem(ContextPropagationUniInterceptor.java:35)
        at io.smallrye.mutiny.operators.UniSerializedSubscriber.onItem(UniSerializedSubscriber.java:72)
        at io.smallrye.mutiny.vertx.AsyncResultUni.lambda$subscribing$1(AsyncResultUni.java:34)
        at io.vertx.kafka.client.producer.impl.KafkaProducerImpl.lambda$send$8(KafkaProducerImpl.java:160)
        at io.vertx.kafka.client.producer.impl.KafkaWriteStreamImpl.lambda$null$1(KafkaWriteStreamImpl.java:118)
        at io.vertx.core.impl.ContextImpl.executeTask(ContextImpl.java:366)
        at io.vertx.core.impl.EventLoopContext.lambda$executeAsync$0(EventLoopContext.java:38)
        at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
        at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500)
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
        at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.lang.Thread.run(Thread.java:748)
```

The application stops to consume new messages and the lag is increasing.

I notice that when I lauch this command : 
```
kafka-console-consumer -bootstrap-server localhost:9092 --topic dead-letter-topic-prices
```
I see an incoming message so the DLQ is correctly populated. 

I use quarkus 1.6

Java : 
λ java -version
java version "1.8.0_261"
Java(TM) SE Runtime Environment (build 1.8.0_261-b12)
Java HotSpot(TM) 64-Bit Server VM (build 25.261-b12, mixed mode)



Changed files:
smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/fault/KafkaDeadLetterQueue.java
smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/fault/KafkaFailureHandlerTest.java

diff --git a/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/fault/KafkaDeadLetterQueue.java b/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/fault/KafkaDeadLetterQueue.java
index 10bb19cbe..07324df42 100644
--- a/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/fault/KafkaDeadLetterQueue.java
+++ b/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/fault/KafkaDeadLetterQueue.java
@@ -72,6 +72,7 @@ public class KafkaDeadLetterQueue implements KafkaFailureHandler {
         log.messageNackedDeadLetter(channel, topic);
         return producer.send(dead)
                 .onFailure().invoke(t -> source.reportFailure((Throwable) t))
+                .onItem().ignore().andContinueWithNull()
                 .subscribeAsCompletionStage();
     }
 }
diff --git a/smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/fault/KafkaFailureHandlerTest.java b/smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/fault/KafkaFailureHandlerTest.java
index dd93c45bf..76eb052af 100644
--- a/smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/fault/KafkaFailureHandlerTest.java
+++ b/smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/fault/KafkaFailureHandlerTest.java
@@ -24,6 +24,7 @@ import org.junit.After;
 import org.junit.Test;
 
 import io.smallrye.config.SmallRyeConfigProviderResolver;
+import io.smallrye.mutiny.Uni;
 import io.smallrye.reactive.messaging.health.HealthReport;
 import io.smallrye.reactive.messaging.kafka.*;
 
@@ -42,7 +43,7 @@ public class KafkaFailureHandlerTest extends KafkaTestBase {
 
     @Test
     public void testFailStrategy() {
-        addConfig(getFailConfig());
+        addConfig(getFailConfig("fail"));
         container = baseWeld().addBeanClass(MyReceiverBean.class).initialize();
 
         await().until(() -> {
@@ -66,9 +67,36 @@ public class KafkaFailureHandlerTest extends KafkaTestBase {
         });
     }
 
+    @Test
+    public void testFailStrategyWithPayload() {
+        addConfig(getFailConfig("fail-payload"));
+        container = baseWeld().addBeanClass(MyReceiverBeanUsingPayload.class).initialize();
+
+        await().until(() -> {
+            HealthReport readiness = getHealth(container).getReadiness();
+            return readiness.isOk();
+        });
+
+        KafkaUsage usage = new KafkaUsage();
+        AtomicInteger counter = new AtomicInteger();
+        new Thread(() -> usage.produceIntegers(10, null,
+                () -> new ProducerRecord<>("fail-payload", counter.getAndIncrement()))).start();
+
+        MyReceiverBeanUsingPayload bean = container.getBeanManager().createInstance().select(MyReceiverBeanUsingPayload.class)
+                .get();
+        await().atMost(2, TimeUnit.MINUTES).until(() -> bean.list().size() >= 4);
+        // Other records should not have been received.
+        assertThat(bean.list()).containsExactly(0, 1, 2, 3);
+
+        await().until(() -> {
+            HealthReport liveness = getHealth(container).getLiveness();
+            return !liveness.isOk();
+        });
+    }
+
     @Test
     public void testIgnoreStrategy() {
-        addConfig(getIgnoreConfig());
+        addConfig(getIgnoreConfig("ignore"));
         container = baseWeld().addBeanClass(MyReceiverBean.class).initialize();
 
         await().until(() -> {
@@ -90,6 +118,31 @@ public class KafkaFailureHandlerTest extends KafkaTestBase {
         assertThat(liveness.isOk()).isTrue();
     }
 
+    @Test
+    public void testIgnoreStrategyWithPayload() {
+        addConfig(getIgnoreConfig("ignore-payload"));
+        container = baseWeld().addBeanClass(MyReceiverBeanUsingPayload.class).initialize();
+
+        await().until(() -> {
+            HealthReport readiness = getHealth(container).getReadiness();
+            return readiness.isOk();
+        });
+
+        KafkaUsage usage = new KafkaUsage();
+        AtomicInteger counter = new AtomicInteger();
+        new Thread(() -> usage.produceIntegers(10, null,
+                () -> new ProducerRecord<>("ignore-payload", counter.getAndIncrement()))).start();
+
+        MyReceiverBeanUsingPayload bean = container.getBeanManager().createInstance().select(MyReceiverBeanUsingPayload.class)
+                .get();
+        await().atMost(2, TimeUnit.MINUTES).until(() -> bean.list().size() >= 10);
+        // All records should not have been received.
+        assertThat(bean.list()).containsExactly(0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
+
+        HealthReport liveness = getHealth(container).getLiveness();
+        assertThat(liveness.isOk()).isTrue();
+    }
+
     @Test
     public void testDeadLetterQueueStrategyWithDefaultTopic() {
         KafkaUsage usage = new KafkaUsage();
@@ -128,6 +181,45 @@ public class KafkaFailureHandlerTest extends KafkaTestBase {
         assertThat(liveness.isOk()).isTrue();
     }
 
+    @Test
+    public void testDeadLetterQueueStrategyWithCustomTopicAndMethodUsingPayload() {
+        KafkaUsage usage = new KafkaUsage();
+        List<ConsumerRecord<String, Integer>> records = new CopyOnWriteArrayList<>();
+        String randomId = UUID.randomUUID().toString();
+
+        usage.consume(randomId, randomId, OffsetResetStrategy.EARLIEST,
+                new StringDeserializer(), new IntegerDeserializer(), () -> records.size() < 3, null, null,
+                Collections.singletonList("dead-letter-topic-kafka-payload"), records::add);
+
+        addConfig(getDeadLetterQueueWithCustomConfig("dq-payload", "dead-letter-topic-kafka-payload"));
+        container = baseWeld().addBeanClass(MyReceiverBeanUsingPayload.class).initialize();
+
+        await().until(() -> {
+            HealthReport readiness = getHealth(container).getReadiness();
+            return readiness.isOk();
+        });
+
+        AtomicInteger counter = new AtomicInteger();
+        new Thread(() -> usage.produceIntegers(10, null,
+                () -> new ProducerRecord<>("dq-payload", counter.getAndIncrement()))).start();
+
+        MyReceiverBeanUsingPayload bean = container.getBeanManager().createInstance().select(MyReceiverBeanUsingPayload.class)
+                .get();
+        await().atMost(2, TimeUnit.MINUTES).until(() -> bean.list().size() >= 10);
+        assertThat(bean.list()).containsExactly(0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
+
+        await().atMost(2, TimeUnit.MINUTES).until(() -> records.size() == 3);
+        assertThat(records).allSatisfy(r -> {
+            assertThat(r.topic()).isEqualTo("dead-letter-topic-kafka-payload");
+            assertThat(r.value()).isIn(3, 6, 9);
+            assertThat(new String(r.headers().lastHeader("dead-letter-reason").value())).startsWith("nack 3 -");
+            assertThat(r.headers().lastHeader("dead-letter-cause")).isNull();
+        });
+
+        HealthReport liveness = getHealth(container).getLiveness();
+        assertThat(liveness.isOk()).isTrue();
+    }
+
     @Test
     public void testDeadLetterQueueStrategyWithCustomConfig() {
         KafkaUsage usage = new KafkaUsage();
@@ -138,7 +230,7 @@ public class KafkaFailureHandlerTest extends KafkaTestBase {
                 new StringDeserializer(), new IntegerDeserializer(), () -> records.size() < 3, null, null,
                 Collections.singletonList("missed"), records::add);
 
-        addConfig(getDeadLetterQueueWithCustomConfig());
+        addConfig(getDeadLetterQueueWithCustomConfig("dead-letter-custom", "missed"));
         container = baseWeld().addBeanClass(MyReceiverBean.class).initialize();
 
         await().until(() -> {
@@ -166,12 +258,12 @@ public class KafkaFailureHandlerTest extends KafkaTestBase {
         assertThat(liveness.isOk()).isTrue();
     }
 
-    private MapBasedConfig getFailConfig() {
+    private MapBasedConfig getFailConfig(String topic) {
         String prefix = "mp.messaging.incoming.kafka.";
         Map<String, Object> config = new HashMap<>();
         config.put(prefix + "connector", KafkaConnector.CONNECTOR_NAME);
         config.put(prefix + "group.id", "my-group");
-        config.put(prefix + "topic", "fail");
+        config.put(prefix + "topic", topic);
         config.put(prefix + "value.deserializer", IntegerDeserializer.class.getName());
         config.put(prefix + "enable.auto.commit", "false");
         config.put(prefix + "auto.offset.reset", "earliest");
@@ -180,11 +272,11 @@ public class KafkaFailureHandlerTest extends KafkaTestBase {
         return new MapBasedConfig(config);
     }
 
-    private MapBasedConfig getIgnoreConfig() {
+    private MapBasedConfig getIgnoreConfig(String topic) {
         String prefix = "mp.messaging.incoming.kafka.";
         Map<String, Object> config = new HashMap<>();
         config.put(prefix + "connector", KafkaConnector.CONNECTOR_NAME);
-        config.put(prefix + "topic", "ignore");
+        config.put(prefix + "topic", topic);
         config.put(prefix + "group.id", "my-group");
         config.put(prefix + "value.deserializer", IntegerDeserializer.class.getName());
         config.put(prefix + "enable.auto.commit", "false");
@@ -208,17 +300,17 @@ public class KafkaFailureHandlerTest extends KafkaTestBase {
         return new MapBasedConfig(config);
     }
 
-    private MapBasedConfig getDeadLetterQueueWithCustomConfig() {
+    private MapBasedConfig getDeadLetterQueueWithCustomConfig(String topic, String dq) {
         String prefix = "mp.messaging.incoming.kafka.";
         Map<String, Object> config = new HashMap<>();
         config.put(prefix + "connector", KafkaConnector.CONNECTOR_NAME);
         config.put(prefix + "group.id", "my-group");
-        config.put(prefix + "topic", "dead-letter-custom");
+        config.put(prefix + "topic", topic);
         config.put(prefix + "value.deserializer", IntegerDeserializer.class.getName());
         config.put(prefix + "enable.auto.commit", "false");
         config.put(prefix + "auto.offset.reset", "earliest");
         config.put(prefix + "failure-strategy", "dead-letter-queue");
-        config.put(prefix + "dead-letter-queue.topic", "missed");
+        config.put(prefix + "dead-letter-queue.topic", dq);
         config.put(prefix + "dead-letter-queue.key.serializer", IntegerSerializer.class.getName());
         config.put(prefix + "dead-letter-queue.value.serializer", IntegerSerializer.class.getName());
 
@@ -227,7 +319,7 @@ public class KafkaFailureHandlerTest extends KafkaTestBase {
 
     @ApplicationScoped
     public static class MyReceiverBean {
-        private List<Integer> received = new ArrayList<>();
+        private final List<Integer> received = new ArrayList<>();
 
         @Incoming("kafka")
         public CompletionStage<Void> process(KafkaRecord<String, Integer> record) {
@@ -244,4 +336,23 @@ public class KafkaFailureHandlerTest extends KafkaTestBase {
         }
 
     }
+
+    @ApplicationScoped
+    public static class MyReceiverBeanUsingPayload {
+        private final List<Integer> received = new ArrayList<>();
+
+        @Incoming("kafka")
+        public Uni<Void> process(int value) {
+            received.add(value);
+            if (value != 0 && value % 3 == 0) {
+                return Uni.createFrom().failure(new IllegalArgumentException("nack 3 - " + value));
+            }
+            return Uni.createFrom().nullItem();
+        }
+
+        public List<Integer> list() {
+            return received;
+        }
+
+    }
 }