Repository: smallrye/smallrye-reactive-messaging
Issue: Unable to use Kafka interceptors while using DLQ

If we configure a channel with a Kafka interceptor and with a DLQ, the DLQ producer fails to create with ` java.lang.ClassCastException: class io.opentracing.contrib.kafka.TracingConsumerInterceptor`.


```
2020-11-09 14:45:48,030 ERROR [io.sma.rea.mes.provider] (Quarkus Main Thread) SRMSG00230: Unable to create the publisher or subscriber during initialization: org.apache.kafka.common.KafkaException: Failed to construct kafka producer
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:434)
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:270)
        at io.vertx.kafka.client.producer.impl.KafkaWriteStreamImpl.create(KafkaWriteStreamImpl.java:52)
        at io.vertx.kafka.client.producer.KafkaWriteStream.create(KafkaWriteStream.java:92)
        at io.vertx.kafka.client.producer.KafkaProducer.create(KafkaProducer.java:149)
        at io.vertx.mutiny.kafka.client.producer.KafkaProducer.create(KafkaProducer.java:142)
        at io.smallrye.reactive.messaging.kafka.fault.KafkaDeadLetterQueue.create(KafkaDeadLetterQueue.java:52)
        at io.smallrye.reactive.messaging.kafka.impl.KafkaSource.createFailureHandler(KafkaSource.java:358)
        at io.smallrye.reactive.messaging.kafka.impl.KafkaSource.<init>(KafkaSource.java:217)
        at io.smallrye.reactive.messaging.kafka.KafkaConnector.getPublisherBuilder(KafkaConnector.java:150)
        at io.smallrye.reactive.messaging.kafka.KafkaConnector_ClientProxy.getPublisherBuilder(KafkaConnector_ClientProxy.zig:305)
        at io.smallrye.reactive.messaging.impl.ConfiguredChannelFactory.createPublisherBuilder(ConfiguredChannelFactory.java:188)
        at io.smallrye.reactive.messaging.impl.ConfiguredChannelFactory.register(ConfiguredChannelFactory.java:153)
        at io.smallrye.reactive.messaging.impl.ConfiguredChannelFactory.initialize(ConfiguredChannelFactory.java:125)
        at io.smallrye.reactive.messaging.impl.ConfiguredChannelFactory_ClientProxy.initialize(ConfiguredChannelFactory_ClientProxy.zig:217)
        at java.base/java.util.Iterator.forEachRemaining(Iterator.java:133)
        at java.base/java.util.Spliterators$IteratorSpliterator.forEachRemaining(Spliterators.java:1801)
        at java.base/java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:658)
        at io.smallrye.reactive.messaging.extension.MediatorManager.initializeAndRun(MediatorManager.java:156)
        at io.smallrye.reactive.messaging.extension.MediatorManager_ClientProxy.initializeAndRun(MediatorManager_ClientProxy.zig:277)
        at io.quarkus.smallrye.reactivemessaging.runtime.SmallRyeReactiveMessagingLifecycle.onApplicationStart(SmallRyeReactiveMessagingLifecycle.java:20)
        at io.quarkus.smallrye.reactivemessaging.runtime.SmallRyeReactiveMessagingLifecycle_Observer_onApplicationStart_4e8937813d9e8faff65c3c07f88fa96615b70e70.notify(SmallRyeReactiveMessagingLifecycle_Observer_onApplicationStart_4e8937813d9e8faff65c3c07f88fa96615b70e70.zig:111)
        at io.quarkus.arc.impl.EventImpl$Notifier.notifyObservers(EventImpl.java:282)
        at io.quarkus.arc.impl.EventImpl$Notifier.notify(EventImpl.java:267)
        at io.quarkus.arc.impl.EventImpl.fire(EventImpl.java:69)
        at io.quarkus.arc.runtime.LifecycleEventRunner.fireStartupEvent(LifecycleEventRunner.java:23)
        at io.quarkus.arc.runtime.ArcRecorder.handleLifecycleEvents(ArcRecorder.java:60)
        at io.quarkus.deployment.steps.LifecycleEventsBuildStep$startupEvent-858218658.deploy_0(LifecycleEventsBuildStep$startupEvent-858218658.zig:81)
        at io.quarkus.deployment.steps.LifecycleEventsBuildStep$startupEvent-858218658.deploy(LifecycleEventsBuildStep$startupEvent-858218658.zig:40)
        at io.quarkus.runner.ApplicationImpl.doStart(ApplicationImpl.zig:488)
        at io.quarkus.runtime.Application.start(Application.java:90)
        at io.quarkus.runtime.ApplicationLifecycleManager.run(ApplicationLifecycleManager.java:95)
        at io.quarkus.runtime.Quarkus.run(Quarkus.java:62)
        at io.quarkus.runtime.Quarkus.run(Quarkus.java:38)
        at io.quarkus.runtime.Quarkus.run(Quarkus.java:104)
        at io.quarkus.runner.GeneratedMain.main(GeneratedMain.zig:29)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:64)
        at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.base/java.lang.reflect.Method.invoke(Method.java:564)
        at io.quarkus.runner.bootstrap.StartupActionImpl$3.run(StartupActionImpl.java:134)
        at java.base/java.lang.Thread.run(Thread.java:832)
Caused by: java.lang.ClassCastException: class io.opentracing.contrib.kafka.TracingConsumerInterceptor
        at java.base/java.lang.Class.asSubclass(Class.java:3851)
        at org.apache.kafka.common.utils.Utils.loadClass(Utils.java:335)
        at org.apache.kafka.common.utils.Utils.newInstance(Utils.java:324)
        at org.apache.kafka.common.config.AbstractConfig.getConfiguredInstance(AbstractConfig.java:365)
        at org.apache.kafka.common.config.AbstractConfig.getConfiguredInstances(AbstractConfig.java:436)
        at org.apache.kafka.common.config.AbstractConfig.getConfiguredInstances(AbstractConfig.java:417)
        at org.apache.kafka.common.config.AbstractConfig.getConfiguredInstances(AbstractConfig.java:404)
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:378)
        ... 41 more
```

Reproducer:
[kafka-interceptor-issue.zip](https://github.com/smallrye/smallrye-reactive-messaging/files/5510879/kafka-interceptor-issue.zip)


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 e69dc3e43..047fa1d00 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
@@ -2,8 +2,7 @@ package io.smallrye.reactive.messaging.kafka.fault;
 
 import static io.smallrye.reactive.messaging.kafka.i18n.KafkaLogging.log;
 import static org.apache.kafka.clients.CommonClientConfigs.CLIENT_ID_CONFIG;
-import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG;
-import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.*;
 import static org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG;
 import static org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG;
 
@@ -49,6 +48,10 @@ public class KafkaDeadLetterQueue implements KafkaFailureHandler {
         Map<String, String> deadQueueProducerConfig = new HashMap<>(kafkaConfiguration);
         String keyDeserializer = deadQueueProducerConfig.remove(KEY_DESERIALIZER_CLASS_CONFIG);
         String valueDeserializer = deadQueueProducerConfig.remove(VALUE_DESERIALIZER_CLASS_CONFIG);
+
+        // We need to remove consumer interceptor
+        deadQueueProducerConfig.remove(INTERCEPTOR_CLASSES_CONFIG);
+
         deadQueueProducerConfig.put(KEY_SERIALIZER_CLASS_CONFIG,
                 conf.getDeadLetterQueueKeySerializer().orElse(getMirrorSerializer(keyDeserializer)));
         deadQueueProducerConfig.put(VALUE_SERIALIZER_CLASS_CONFIG,
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 4fb8b4cc2..9b49c4a49 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
@@ -4,10 +4,7 @@ import static io.smallrye.reactive.messaging.kafka.fault.KafkaDeadLetterQueue.*;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
 
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.UUID;
+import java.util.*;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
@@ -18,11 +15,10 @@ import java.util.function.Function;
 import javax.enterprise.context.ApplicationScoped;
 import javax.enterprise.event.Observes;
 
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.OffsetResetStrategy;
+import org.apache.kafka.clients.consumer.*;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.serialization.IntegerDeserializer;
 import org.apache.kafka.common.serialization.IntegerSerializer;
 import org.apache.kafka.common.serialization.StringDeserializer;
@@ -229,6 +225,45 @@ public class KafkaFailureHandlerTest extends KafkaTestBase {
         assertThat(bean.producers()).isEqualTo(1);
     }
 
+    @Test
+    public void testDeadLetterQueueStrategyWithInterceptor() {
+        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-itcp"), records::add);
+
+        MyReceiverBeanUsingPayload bean = runApplication(
+                getDeadLetterQueueWithCustomConfig("dq-itcp", "dead-letter-topic-kafka-itcp")
+                        .with("mp.messaging.incoming.kafka.interceptor.classes", IdentityInterceptor.class.getName()),
+                MyReceiverBeanUsingPayload.class);
+        await().until(this::isReady);
+
+        AtomicInteger counter = new AtomicInteger();
+        new Thread(() -> usage.produceIntegers(10, null,
+                () -> new ProducerRecord<>("dq-itcp", counter.getAndIncrement()))).start();
+
+        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-itcp");
+            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();
+            assertThat(new String(r.headers().lastHeader(DEAD_LETTER_PARTITION).value())).isEqualTo("0");
+            assertThat(new String(r.headers().lastHeader(DEAD_LETTER_TOPIC).value())).isEqualTo("dq-itcp");
+            assertThat(new String(r.headers().lastHeader(DEAD_LETTER_OFFSET).value())).isNotNull().isIn("3", "6", "9");
+        });
+
+        assertThat(isAlive()).isTrue();
+
+        assertThat(bean.consumers()).isEqualTo(1);
+        assertThat(bean.producers()).isEqualTo(1);
+    }
+
     @Test
     public void testDeadLetterQueueStrategyWithCustomConfig() {
         List<ConsumerRecord<String, Integer>> records = new CopyOnWriteArrayList<>();
@@ -397,4 +432,28 @@ public class KafkaFailureHandlerTest extends KafkaTestBase {
             return observedProducerEvents.sum();
         }
     }
+
+    public static class IdentityInterceptor<K, V> implements ConsumerInterceptor<K, V> {
+        @Override
+        public ConsumerRecords<K, V> onConsume(
+                ConsumerRecords<K, V> records) {
+            return records;
+        }
+
+        @Override
+        public void onCommit(
+                Map<TopicPartition, OffsetAndMetadata> offsets) {
+
+        }
+
+        @Override
+        public void close() {
+
+        }
+
+        @Override
+        public void configure(Map<String, ?> configs) {
+
+        }
+    }
 }