fix(impl): Skip ProducerOnSendInterceptor execution when publishing to the local DLQ

This PR resolves an inconsistency in DLQ handling where processor failures were sent to the LocalDLQ using the topology-defined Kafka producer, causing ProducerOnSendInterceptor logic to be applied and potentially altering the original poisonous message.

The refactoring introduces a dedicated raw Kafka producer, explicitly marked for DLQ usage, which bypasses interceptor execution.
This producer is now shared between DlqDecorator (for processor failures) and LogAndSendToDlqExceptionHandlerDelegate (for deserialization exceptions), ensuring all LocalDLQ records are pushed unmodified except for DLQ metadata headers, and providing consistent DLQ behavior across all failure scenarios.

Fixes: #233

Author: lmartella1

impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/TopologyProducer.java

@@ -29,14 +29,12 @@
 import jakarta.inject.Provider;
 
 import org.apache.kafka.common.serialization.Deserializer;
-import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.streams.KafkaClientSupplier;
 import org.apache.kafka.streams.Topology;
 import org.apache.kafka.streams.processor.api.Processor;
 import org.apache.kafka.streams.processor.api.ProcessorSupplier;
 
 import io.cloudevents.kafka.CloudEventDeserializer;
-import io.cloudevents.kafka.CloudEventSerializer;
 import io.quarkiverse.kafkastreamsprocessor.api.configuration.ConfigurationCustomizer;
 import io.quarkiverse.kafkastreamsprocessor.api.decorator.producer.ProducerOnSendInterceptor;
 import io.quarkiverse.kafkastreamsprocessor.impl.configuration.DefaultConfigurationCustomizer;
@@ -59,11 +57,6 @@ public class TopologyProducer {
      */
     public static final String PROCESSOR_NAME = "Processor";
 
-    /**
-     * Default sink name of the dead-letter queue
-     */
-    public static final String DLQ_SINK_NAME = "DLQ";
-
     /**
      * Class containing the configuration related to kafka streams processor
      */
@@ -194,11 +187,6 @@ public Topology topology(TopologyConfigurationImpl configuration,
         sinkToTopicMapping
                 .forEach((String sink, String topic) -> topology.addSink(sink, topic, configuration.getSinkKeySerializer(),
                         configuration.getSinkValueSerializer(), PROCESSOR_NAME));
-        if (kStreamsProcessorConfig.dlq().topic().isPresent()) {
-            topology.addSink(DLQ_SINK_NAME, kStreamsProcessorConfig.dlq().topic().get(),
-                    configuration.getSourceKeySerde().serializer(),
-                    getSourceValueSerializer(configuration, kStreamsProcessorConfig), PROCESSOR_NAME);
-        }
 
         configuration.getStoreConfigurations()
                 .forEach(storeConfiguration -> topology.addStateStore(storeConfiguration.getStoreBuilder(), PROCESSOR_NAME));
@@ -218,17 +206,6 @@ private Deserializer<?> getSourceValueDeserializer(TopologyConfigurationImpl con
         return configuration.getSourceValueSerde().deserializer();
     }
 
-    private Serializer<?> getSourceValueSerializer(TopologyConfigurationImpl configuration,
-            KStreamsProcessorConfig kStreamsProcessorConfig) {
-        // defaulting to CloudEventSerializer if kafkastreamsprocessor.input.is-cloud-event is true
-        if (kStreamsProcessorConfig.input().isCloudEvent()) {
-            CloudEventSerializer serializer = new CloudEventSerializer();
-            serializer.configure(kStreamsProcessorConfig.dlq().cloudEventSerializerConfig(), false);
-            return serializer;
-        }
-        return configuration.getSourceValueSerde().serializer();
-    }
-
     private void addGlobalStores(TopologyConfigurationImpl configuration, Topology topology) {
         configuration.getGlobalStoreConfigurations()
                 .forEach(config -> {

impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/DlqDecorator.java

@@ -19,6 +19,8 @@
  */
 package io.quarkiverse.kafkastreamsprocessor.impl.decorator.processor;
 
+import static org.apache.kafka.common.record.TimestampType.CREATE_TIME;
+
 import java.util.Optional;
 import java.util.Set;
 
@@ -27,7 +29,9 @@
 import jakarta.enterprise.context.Dependent;
 import jakarta.inject.Inject;
 
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.streams.processor.To;
 import org.apache.kafka.streams.processor.api.ContextualProcessor;
 import org.apache.kafka.streams.processor.api.FixedKeyRecord;
@@ -37,12 +41,12 @@
 import org.apache.kafka.streams.processor.api.RecordMetadata;
 import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
 
+import io.cloudevents.kafka.CloudEventSerializer;
+import io.quarkiverse.kafkastreamsprocessor.api.configuration.Configuration;
 import io.quarkiverse.kafkastreamsprocessor.api.decorator.processor.AbstractProcessorDecorator;
 import io.quarkiverse.kafkastreamsprocessor.api.decorator.processor.ProcessorDecoratorPriorities;
-import io.quarkiverse.kafkastreamsprocessor.impl.TopologyProducer;
-import io.quarkiverse.kafkastreamsprocessor.impl.errors.DlqMetadataHandler;
+import io.quarkiverse.kafkastreamsprocessor.impl.errors.DlqProducerService;
 import io.quarkiverse.kafkastreamsprocessor.impl.errors.ErrorHandlingStrategy;
-import io.quarkiverse.kafkastreamsprocessor.impl.metrics.KafkaStreamsProcessorMetrics;
 import io.quarkiverse.kafkastreamsprocessor.spi.SinkToTopicMappingBuilder;
 import io.quarkiverse.kafkastreamsprocessor.spi.properties.KStreamsProcessorConfig;
 import lombok.AccessLevel;
@@ -68,56 +72,66 @@ public class DlqDecorator extends AbstractProcessorDecorator {
     private final Set<String> functionalSinks;
 
     /**
-     * Tool to enrich a message metadata before its storage in the dead letter queue
+     * Whether the dead-letter queue mechanism is activated for this microservice
      */
-    private final DlqMetadataHandler dlqMetadataHandler;
+    private final boolean activated;
 
     /**
-     * container of all metrics of the framework
+     * Delegate that handles DLQ message production
      */
-    private final KafkaStreamsProcessorMetrics metrics;
+    private final DlqProducerService dlqDelegate;
 
     /**
-     * Whether the dead-letter queue mechanism is activated for this microservice
+     * The configuration of the Kafka Streams processor
      */
-    private final boolean activated;
+    private final Configuration configuration;
+
+    /**
+     * Class containing the configuration related to kafka streams processor
+     */
+    private final KStreamsProcessorConfig kStreamsProcessorConfig;
 
     /**
      * Keeping a reference to the ProcessorContext to be able to use it in the {@link Processor#process(Record)} method
      * whilst not implementing the little too narrowing {@link ContextualProcessor}.
      */
     private ProcessorContext context;
 
-    DlqDecorator(Set<String> functionalSinks, DlqMetadataHandler dlqMetadataHandler,
-            KafkaStreamsProcessorMetrics metrics, boolean activated) {
+    DlqDecorator(Set<String> functionalSinks, boolean activated,
+            DlqProducerService dlqDelegate,
+            Configuration configuration,
+            KStreamsProcessorConfig kStreamsProcessorConfig) {
         this.functionalSinks = functionalSinks;
-        this.dlqMetadataHandler = dlqMetadataHandler;
-        this.metrics = metrics;
         this.activated = activated;
+        this.dlqDelegate = dlqDelegate;
+        this.configuration = configuration;
+        this.kStreamsProcessorConfig = kStreamsProcessorConfig;
     }
 
     /**
      * Injection constructor
      *
      * @param sinkToTopicMappingBuilder
      *        utility to get access to the mapping between sinks and Kafka topics
-     * @param dlqMetadataHandler
-     *        the enricher of metadata before sending message to the dead letter queue
-     * @param metrics
-     *        container of all metrics of the framework
      * @param kStreamsProcessorConfig
      *        It contains the configuration for the error strategy configuration property value (default
      *        {@link ErrorHandlingStrategy#CONTINUE})
      *        and the configuration Kafka topic to use for dead letter queue (optional)
+     * @param dlqDelegate
+     *        delegate that handles DLQ production with properly configured producer
+     * @param configuration
+     *        the configuration of the kafka streams processor
      */
     @Inject
     public DlqDecorator(
-            SinkToTopicMappingBuilder sinkToTopicMappingBuilder, DlqMetadataHandler dlqMetadataHandler,
-            KafkaStreamsProcessorMetrics metrics,
-            KStreamsProcessorConfig kStreamsProcessorConfig) { // NOSONAR Optional with microprofile-config
-        this(sinkToTopicMappingBuilder.sinkToTopicMapping().keySet(), dlqMetadataHandler, metrics,
+            SinkToTopicMappingBuilder sinkToTopicMappingBuilder,
+            KStreamsProcessorConfig kStreamsProcessorConfig,
+            DlqProducerService dlqDelegate,
+            Configuration configuration) { // NOSONAR Optional with microprofile-config
+        this(sinkToTopicMappingBuilder.sinkToTopicMapping().keySet(),
                 ErrorHandlingStrategy.shouldSendToDlq(kStreamsProcessorConfig.errorStrategy(),
-                        kStreamsProcessorConfig.dlq().topic()));
+                        kStreamsProcessorConfig.dlq().topic()),
+                dlqDelegate, configuration, kStreamsProcessorConfig);
     }
 
     /**
@@ -158,11 +172,30 @@ public void process(Record record) {
             } catch (RuntimeException e) { // NOSONAR
                 Optional<RecordMetadata> recordMetadata = context.recordMetadata();
                 if (recordMetadata.isPresent()) {
-                    dlqMetadataHandler.addMetadata(record.headers(), recordMetadata.get().topic(),
-                            recordMetadata.get().partition(), e);
-                    context.forward(record, TopologyProducer.DLQ_SINK_NAME);
-                    // Re-throw so the exception gets logged
-                    metrics.microserviceDlqSentCounter().increment();
+                    Serializer<Object> keySerializer = (Serializer<Object>) configuration.getSourceKeySerde().serializer();
+                    Serializer<Object> valueSerializer = (Serializer<Object>) getSourceValueSerializer(configuration,
+                            kStreamsProcessorConfig);
+                    byte[] serializedKey = keySerializer.serialize(recordMetadata.get().topic(), record.headers(),
+                            record.key());
+                    byte[] serializedValue = valueSerializer.serialize(recordMetadata.get().topic(), record.headers(),
+                            record.value());
+
+                    ConsumerRecord<byte[], byte[]> consumerRecord = new ConsumerRecord<>(
+                            recordMetadata.get().topic(),
+                            recordMetadata.get().partition(),
+                            recordMetadata.get().offset(),
+                            record.timestamp(),
+                            CREATE_TIME,
+                            serializedKey != null ? serializedKey.length : 0,
+                            serializedValue != null ? serializedValue.length : 0,
+                            serializedKey,
+                            serializedValue,
+                            record.headers(),
+                            Optional.empty());
+
+                    dlqDelegate.sendToDlq(consumerRecord, e, context.taskId(), false);
+
+                    // throw exception to fail OpenTelemetry span
                     throw e;
                 }
             }
@@ -171,6 +204,17 @@ public void process(Record record) {
         }
     }
 
+    Serializer<?> getSourceValueSerializer(Configuration configuration,
+            KStreamsProcessorConfig kStreamsProcessorConfig) {
+        // defaulting to CloudEventSerializer if kafkastreamsprocessor.input.is-cloud-event is true
+        if (kStreamsProcessorConfig.input().isCloudEvent()) {
+            CloudEventSerializer serializer = new CloudEventSerializer();
+            serializer.configure(kStreamsProcessorConfig.dlq().cloudEventSerializerConfig(), false);
+            return serializer;
+        }
+        return configuration.getSourceValueSerde().serializer();
+    }
+
     @RequiredArgsConstructor(access = AccessLevel.MODULE)
     static final class DlqProcessorContextDecorator<KOut, VOut> implements InternalProcessorContext<KOut, VOut> {
 

impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/errors/DlqProducerService.java

@@ -0,0 +1,140 @@
+/*-
+ * #%L
+ * Quarkus Kafka Streams Processor
+ * %%
+ * Copyright (C) 2024 Amadeus s.a.s.
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+package io.quarkiverse.kafkastreamsprocessor.impl.errors;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import jakarta.annotation.PreDestroy;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.Configurable;
+import org.apache.kafka.streams.KafkaClientSupplier;
+import org.apache.kafka.streams.processor.TaskId;
+
+import io.quarkiverse.kafkastreamsprocessor.api.decorator.producer.ProducerOnSendInterceptor;
+import io.quarkiverse.kafkastreamsprocessor.impl.KafkaClientSupplierDecorator;
+import io.quarkiverse.kafkastreamsprocessor.impl.metrics.KafkaStreamsProcessorMetrics;
+import io.quarkiverse.kafkastreamsprocessor.spi.properties.KStreamsProcessorConfig;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Service responsible for producing records to the Dead Letter Queue (DLQ) topic in Kafka Streams Processor.
+ * <p>
+ * This class manages the lifecycle and configuration of the DLQ producer, enriches records with metadata,
+ * and tracks metrics for DLQ operations.
+ */
+@Slf4j
+@ApplicationScoped
+public class DlqProducerService implements Configurable {
+
+    /**
+     * Kafka producer supplier for this framework
+     */
+    private final KafkaClientSupplier clientSupplier;
+
+    /**
+     * Metrics container for this framework
+     */
+    private final KafkaStreamsProcessorMetrics metrics;
+
+    /**
+     * Tool object that enriches the metadata of messages before sending them to the microservice's specific DLQ.
+     */
+    private final DlqMetadataHandler dlqMetadataHandler;
+
+    /**
+     * The class containing all the configuration related to kafka stream processor
+     */
+    private final KStreamsProcessorConfig kStreamsProcessorConfig;
+
+    /** Producer for the dlq topic */
+    private Producer<byte[], byte[]> dlqProducer;
+
+    /** True if the dead letter queue strategy is selected and properly configured */
+    boolean sendToDlq;
+
+    /**
+     * Injection constructor
+     *
+     * @param kafkaClientSupplier a supplier of {@link org.apache.kafka.clients.producer.KafkaProducer}
+     * @param metrics the metrics container of this framework
+     * @param dlqMetadataHandler tool to enrich message metadata before sending them to the microservice's DLQ
+     *        the configuration error strategy for the application. See { @link {@link ErrorHandlingStrategy}
+     * @param kStreamsProcessorConfig The configuration related to kafka processor
+     */
+    @Inject
+    public DlqProducerService(KafkaClientSupplier kafkaClientSupplier,
+            KafkaStreamsProcessorMetrics metrics,
+            DlqMetadataHandler dlqMetadataHandler,
+            KStreamsProcessorConfig kStreamsProcessorConfig) {
+        this.clientSupplier = kafkaClientSupplier;
+        this.metrics = metrics;
+        this.dlqMetadataHandler = dlqMetadataHandler;
+        this.kStreamsProcessorConfig = kStreamsProcessorConfig;
+    }
+
+    public void sendToDlq(final ConsumerRecord<byte[], byte[]> record, final Exception exception, TaskId taskId,
+            Boolean isDeserializationException) {
+        metrics.microserviceDlqSentCounter().increment();
+        log.error("Exception caught during {}, sending to the dead letter queue topic; " +
+                "taskId: {}, topic: {}, partition: {}, offset: {}",
+                isDeserializationException ? "deserialization" : "processing",
+                taskId, record.topic(), record.partition(), record.offset(),
+                exception);
+
+        dlqProducer.send(new ProducerRecord<>(kStreamsProcessorConfig.dlq().topic().get(), null, record.timestamp(),
+                record.key(), record.value(),
+                dlqMetadataHandler.withMetadata(record.headers(), record.topic(), record.partition(), exception)));
+    }
+
+    /**
+     * If DLQ is active, it initializes a {@link Producer} for the DLQ with the
+     * {@link KafkaClientSupplierDecorator#DLQ_PRODUCER} flag
+     * so it is only decorated with {@link ProducerOnSendInterceptor} that have
+     * {@link ProducerOnSendInterceptor#skipForDLQ()} returning <code>false</code>.
+     * <p>
+     * <b>Original documentation:</b>
+     * <p>
+     * {@inheritDoc}
+     */
+    @Override
+    public void configure(final Map<String, ?> configs) {
+        // Resolve the DLQ strategy once to fail fast in case of misconfiguration
+        sendToDlq = ErrorHandlingStrategy.shouldSendToDlq(kStreamsProcessorConfig.errorStrategy(),
+                kStreamsProcessorConfig.dlq().topic());
+        if (sendToDlq) {
+            Map<String, Object> dlqConfigMap = new HashMap<>(configs);
+            dlqConfigMap.put(KafkaClientSupplierDecorator.DLQ_PRODUCER, true);
+            dlqProducer = new LogCallbackExceptionProducerDecorator(clientSupplier.getProducer(dlqConfigMap));
+        }
+    }
+
+    @PreDestroy
+    void close() {
+        if (dlqProducer != null) {
+            dlqProducer.close(GlobalDLQProductionExceptionHandlerDelegate.GRACEFUL_PERIOD);
+        }
+    }
+}