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>