feat(impl): Support interception of output record before serialization

Fixes #147.

Introduction of a new `OutputRecordInterceptor` interface to be implemented to be able to intercept outgoing messages before their payload is serialized.
Its aim is to replace the `ProducerOnSendInterceptor` that at the difference with `OutputRecordInterceptor` is called after the serialization of the payload.
Henceforth `ProducerOnSendInterceptor` is deprecated.

Version is bumped to 4.1 to reflect the new feature.
It is backwards-compatible.

The TracingProducerInterceptor has been rewritten as a TracingOutputRecordInterceptor, but kept for backwards-compatibility.
It is itself marked for removal as well.

Documentation updated accordingly.

Author: edeweerd1A

.github/project.yml

@@ -1,3 +1,3 @@
 release:
   current-version: "4.0.3"
-  next-version: "4.0.0-SNAPSHOT"
+  next-version: "4.1.0-SNAPSHOT"

api/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>io.quarkiverse.kafkastreamsprocessor</groupId>
     <artifactId>quarkus-kafka-streams-processor-parent</artifactId>
-    <version>4.0.0-SNAPSHOT</version>
+    <version>4.1.0-SNAPSHOT</version>
   </parent>
   <artifactId>quarkus-kafka-streams-processor-api</artifactId>
   <name>quarkus-kafka-streams-processor-api</name>

api/src/main/java/io/quarkiverse/kafkastreamsprocessor/api/decorator/outputrecord/OutputRecordInterceptor.java

@@ -0,0 +1,39 @@
+package io.quarkiverse.kafkastreamsprocessor.api.decorator.outputrecord;
+
+import org.apache.kafka.streams.processor.api.Record;
+
+/**
+ * Interceptor that if implemented is called whenever the processor calls any of
+ * {@link org.apache.kafka.streams.processor.api.ProcessorContext}'s forward methods.
+ * <p>
+ * Order of execution is guaranteed based on the integer priority returned by {@link #priority()}.
+ * </p>
+ * <p>
+ * It differs from {@link io.quarkiverse.kafkastreamsprocessor.api.decorator.producer.ProducerOnSendInterceptor} in that
+ * it is executed before any serialization of the payload to byte[] happens.
+ * </p>
+ */
+public interface OutputRecordInterceptor {
+    /**
+     * By default, if not overriden, the interceptor has the following priority.
+     */
+    int DEFAULT_PRIORITY = 100;
+
+    /**
+     * Override this method to finely tune the order of execution of any interceptor you implement.
+     *
+     * @return the custom priority you want to assign. A number between 0 and {@link Integer#MAX_VALUE}.
+     */
+    default int priority() {
+        return DEFAULT_PRIORITY;
+    }
+
+    /**
+     * Intercept the record before it is eventually given to
+     * {@link org.apache.kafka.streams.processor.api.ProcessorContext#forward(Record, String)}.
+     *
+     * @param record the record has the processor requested it to be forwarded
+     * @return the new record with any modifications this interceptor wants to apply before serialization.
+     */
+    Record interceptOutputRecord(Record record);
+}

api/src/main/java/io/quarkiverse/kafkastreamsprocessor/api/decorator/outputrecord/OutputRecordInterceptorPriorities.java

@@ -0,0 +1,34 @@
+/*-
+ * #%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.api.decorator.outputrecord;
+
+/**
+ * Priorities of the output record interceptors the framework provides.
+ */
+public final class OutputRecordInterceptorPriorities {
+    /**
+     * Priority of the interceptor that will inject the tracing headers for propagation.
+     */
+    public static final int TRACING = 100;
+
+    private OutputRecordInterceptorPriorities() {
+
+    }
+}

api/src/main/java/io/quarkiverse/kafkastreamsprocessor/api/decorator/processor/AbstractProcessorDecorator.java

@@ -39,11 +39,7 @@
  * Class introduced in 2.0, for compatibility with Quarkus 3.8 random failure to start when using custom processor
  * decorators.
  * </p>
- *
- * @deprecated It will be removed in 3.0, with the integration of Quarkus 3.15 where we will be able to go back to pure
- *             CDI decorators.
  */
-@Deprecated(forRemoval = true, since = "2.0")
 public abstract class AbstractProcessorDecorator implements Processor {
     /**
      * The decorated processor, holding either the next decorator layer or the final processor.

api/src/main/java/io/quarkiverse/kafkastreamsprocessor/api/decorator/processor/ProcessorDecoratorPriorities.java

@@ -58,25 +58,36 @@ public final class ProcessorDecoratorPriorities {
      * {@link ContextualProcessor#process(Record)} method.
      */
     public static final int TRACING = 100;
+
     /**
      * Priority of the decorator in charge or initializing a "request context" for the duration of the processing of the
      * ContextualProcessor#process(Record)} method. It is closed afterward.
      */
     public static final int CDI_REQUEST_SCOPE = 200;
+
+    /**
+     * Priority for the decorator that wraps the {@link org.apache.kafka.streams.processor.api.ProcessorContext} to
+     * intercept calls to its <code>forward</code> methods.
+     */
+    public static final int CONTEXT_FORWARD = 250;
+
     /**
      * Priority of the decorator that will handle exception and potentially redirect the message in a dead letter queue
      * topic, if configured.
      */
     public static final int DLQ = 300;
+
     /**
      * Priority of the decorator in charge of measuring the processing time and the number of exceptions thrown.
      */
     public static final int METRICS = 400;
+
     /**
      * Priority of the decorator in charge of injecting all {@link DecoratedPunctuator} configured by the framework and
      * your custom potential additions.
      */
     public static final int PUNCTUATOR_DECORATION = 500;
+
     /**
      * Priority of the decorator in charge of implementing a form of fault tolerance by means of calling again the
      * {@link ContextualProcessor#process(Record)} method.

api/src/main/java/io/quarkiverse/kafkastreamsprocessor/api/decorator/producer/ProducerInterceptorPriorities.java

@@ -21,7 +21,11 @@
 
 /**
  * Priorities of the producer interceptors the framework provides.
+ *
+ * @deprecated Change any producer interceptor into a
+ *             {@link io.quarkiverse.kafkastreamsprocessor.api.decorator.outputrecord.OutputRecordInterceptor}.
  */
+@Deprecated(forRemoval = true, since = "4.1")
 public final class ProducerInterceptorPriorities {
     /**
      * Priority of the interceptor that will inject the tracing headers for propagation.

api/src/main/java/io/quarkiverse/kafkastreamsprocessor/api/decorator/producer/ProducerOnSendInterceptor.java

@@ -26,7 +26,11 @@
 
 /**
  * Interface to extend to by able to decorate the production of the response message to the outgoing topic.
+ *
+ * @deprecated Change any producer interceptor into a
+ *             {@link io.quarkiverse.kafkastreamsprocessor.api.decorator.outputrecord.OutputRecordInterceptor}.
  */
+@Deprecated(forRemoval = true, since = "4.1")
 public interface ProducerOnSendInterceptor extends ProducerInterceptor<byte[], byte[]> {
     /**
      * By default, if not overriden, the interceptor has the following priority.

bom/application/pom.xml

@@ -4,13 +4,13 @@
   <parent>
     <groupId>io.quarkiverse.kafkastreamsprocessor</groupId>
     <artifactId>quarkus-kafka-streams-processor-bom-parent</artifactId>
-    <version>4.0.0-SNAPSHOT</version>
+    <version>4.1.0-SNAPSHOT</version>
   </parent>
   <artifactId>quarkus-kafka-streams-processor-bom</artifactId>
   <name>quarkus-kafka-streams-processor-bom</name>
   <packaging>pom</packaging>
   <properties>
-    <quarkus-kafka-streams-processor.version>4.0.0-SNAPSHOT</quarkus-kafka-streams-processor.version>
+    <quarkus-kafka-streams-processor.version>4.1.0-SNAPSHOT</quarkus-kafka-streams-processor.version>
   </properties>
   <dependencyManagement>
     <dependencies>

bom/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>io.quarkiverse.kafkastreamsprocessor</groupId>
     <artifactId>quarkus-kafka-streams-processor-parent</artifactId>
-    <version>4.0.0-SNAPSHOT</version>
+    <version>4.1.0-SNAPSHOT</version>
   </parent>
   <artifactId>quarkus-kafka-streams-processor-bom-parent</artifactId>
   <name>quarkus-kafka-streams-processor-bom-parent</name>

bom/test/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>io.quarkiverse.kafkastreamsprocessor</groupId>
     <artifactId>quarkus-kafka-streams-processor-bom-parent</artifactId>
-    <version>4.0.0-SNAPSHOT</version>
+    <version>4.1.0-SNAPSHOT</version>
   </parent>
   <artifactId>quarkus-kafka-streams-processor-test-bom</artifactId>
   <name>quarkus-kafka-streams-processor-test-bom</name>
@@ -19,7 +19,7 @@
       <dependency>
         <groupId>org.hamcrest</groupId>
         <artifactId>hamcrest</artifactId>
-        <version>2.2</version>
+        <version>3.0</version>
       </dependency>
       <dependency>
         <groupId>com.spotify</groupId>
@@ -89,7 +89,7 @@
       <dependency>
         <groupId>org.assertj</groupId>
         <artifactId>assertj-core</artifactId>
-        <version>3.26.3</version>
+        <version>3.27.3</version>
       </dependency>
       <!-- Mockserver -->
       <dependency>

deployment/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>io.quarkiverse.kafkastreamsprocessor</groupId>
     <artifactId>quarkus-kafka-streams-processor-parent</artifactId>
-    <version>4.0.0-SNAPSHOT</version>
+    <version>4.1.0-SNAPSHOT</version>
   </parent>
   <artifactId>quarkus-kafka-streams-processor-deployment</artifactId>
   <name>quarkus-kafka-streams-processor-deployment</name>

docs/modules/ROOT/pages/index.adoc

@@ -120,7 +120,7 @@ For keys, only type `String` is supported.
 For values, see <<_data_serialization, supported types>>.
 <3> The `process` method is called for each incoming message.
 Any processing must be done synchronously.
-<4> The `forward` method allows to produce an outgoing message.
+<4> The `outputrecord` method allows to produce an outgoing message.
 Internally Kafka Streams links the acknowledgment on the outgoing message with the commit of the associated incoming message.
 
 === Topology builder
@@ -745,6 +745,11 @@ The priority is to be set based on the priorities of the existing decorators whi
      * your custom potential additions.
      */
     public static final int PUNCTUATOR_DECORATION = 500;
+    /**
+     * Priority for the decorator that wraps the {@link org.apache.kafka.streams.processor.api.ProcessorContext} to
+     * intercept calls to its <code>forward</code> methods.
+     */
+    public static final int CONTEXT_FORWARD = 550;
     /**
      * Priority of the decorator in charge of implementing a form of fault tolerance by means of calling again the
      * {@link ContextualProcessor#process(Record)} method.
@@ -757,7 +762,52 @@ The priority is to be set based on the priorities of the existing decorators whi
 Such a decorator will automatically been taken into account by CDI.
 The priority will control at which point your decorator will be called among all other decorators.
 
-=== Producer interceptor
+[#_output_record_interceptor]
+=== Output record interceptor
+
+A `Processor` usually ends with the forwarding of an outgoing message.
+For that purpose, the `ProcessorContext#foward` methods are to be used.
+
+From `quarkus-kafka-streams-processor` 4.1, calls to those methods can be intercepted.
+If you define a bean of type `OutputRecordInterceptor`, it will be taken up by the framework and executed.
+There is an order to multiple instances of interceptors, driven by the value returned by `OutputRecordInterceptor#priority`.
+
+Example of interceptor:
+
+[source,java]
+----
+@ApplicationScoped
+public class MyOutputRecordInterceptor implements OutputRecordInterceptor {
+    @Override
+    public Record interceptOutputRecord(Record record) {
+        record.headers().add("MyHeader", "MyValue".getBytes(StandardCharsets.UTF_8));
+        return record;
+    }
+
+    @Override
+    public int priority() {
+        return 150; // <1>
+    }
+}
+----
+
+<1> Priority of this interceptor.
+Default priority is 100.
+References for priorities of existing interceptors this library provides:
++
+[source,java]
+----
+    /**
+     * Priority of the interceptor that will inject the tracing headers for propagation.
+     */
+    public static final int TRACING = 100;
+----
++
+Lower priority are executed first.
+
+=== Producer interceptor [deprecated]
+
+WARNING: Producer interceptors are from 4.1 deprecated. They should be reimplemented as xref:_output_record_interceptor[OutputRecordInterceptor].
 
 Kafka Streams already has the concept of a https://kafka.apache.org/36/javadoc/org/apache/kafka/clients/producer/ProducerInterceptor.html[ProducerInterceptor].
 But as the rest of Kafka Streams SPI, it is based on a class name and a default constructor for instantiation.

docs/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>io.quarkiverse.kafkastreamsprocessor</groupId>
         <artifactId>quarkus-kafka-streams-processor-parent</artifactId>
-        <version>4.0.0-SNAPSHOT</version>
+        <version>4.1.0-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

impl/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>io.quarkiverse.kafkastreamsprocessor</groupId>
     <artifactId>quarkus-kafka-streams-processor-parent</artifactId>
-    <version>4.0.0-SNAPSHOT</version>
+    <version>4.1.0-SNAPSHOT</version>
   </parent>
   <artifactId>quarkus-kafka-streams-processor-impl</artifactId>
   <name>quarkus-kafka-streams-processor-impl</name>

impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/outputrecord/TracingOutputRecordInterceptor.java

@@ -0,0 +1,61 @@
+/*-
+ * #%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.decorator.outputrecord;
+
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+
+import org.apache.kafka.streams.processor.api.Record;
+
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.context.Context;
+import io.quarkiverse.kafkastreamsprocessor.api.decorator.outputrecord.OutputRecordInterceptor;
+import io.quarkiverse.kafkastreamsprocessor.api.decorator.outputrecord.OutputRecordInterceptorPriorities;
+import io.quarkiverse.kafkastreamsprocessor.propagation.KafkaTextMapSetter;
+
+/**
+ * Producer interceptor that injects the tracing headers for propagation.
+ */
+@ApplicationScoped
+public class TracingOutputRecordInterceptor implements OutputRecordInterceptor {
+    private final OpenTelemetry openTelemetry;
+
+    private final KafkaTextMapSetter kafkaTextMapSetter;
+
+    @Inject
+    public TracingOutputRecordInterceptor(OpenTelemetry openTelemetry, KafkaTextMapSetter kafkaTextMapSetter) {
+        this.openTelemetry = openTelemetry;
+        this.kafkaTextMapSetter = kafkaTextMapSetter;
+    }
+
+    @Override
+    public Record interceptOutputRecord(Record record) {
+        openTelemetry.getPropagators().getTextMapPropagator().fields().forEach(record.headers()::remove);
+        openTelemetry.getPropagators()
+                .getTextMapPropagator()
+                .inject(Context.current(), record.headers(), kafkaTextMapSetter);
+        return record;
+    }
+
+    @Override
+    public int priority() {
+        return OutputRecordInterceptorPriorities.TRACING;
+    }
+}