Add coverage for ArC fix when a framework bean uses the decorator

This adding the coverage for https://issues.redhat.com/browse/QUARKUS-5178
It's tested with Quarkiverse extension as the reproducer attached in issue
and it was only scenario which I was able to reproduce.

To test this you need external lib which contains class (lib_class) implementing the interface.
This interface is also aplied on class anotated as decorator in Quarkus app when the bean of lib_class is injected at the same time.

This error happen randomly in dev/test mode.

The Quarkus PR is quarkusio/quarkus#43245

README.md

@@ -974,6 +974,11 @@ There is an EventsProducer that generate stock prices events every 1s. The event
 A Kafka consumer will read these events serialized by AVRO and change an `status` property to `COMPLETED`.
 The streams of completed events will be exposed through an SSE endpoint.
 
+### `messaging/kafka-processor`
+
+This module verify the [QUARKUS-5178](https://issues.redhat.com/browse/QUARKUS-5178), which using Quarkiverse extension.
+This is tested only in dev mode only as it's not happening in prod mode.
+
 ### `messaging/kafka-strimzi-avro-reactive-messaging`
 
 - Verifies that `Quarkus Kafka` + `Apicurio Kakfa Registry`(AVRO) and `Quarkus SmallRye Reactive Messaging` extensions work as expected.

messaging/kafka-processor/pom.xml

@@ -0,0 +1,33 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>io.quarkus.ts.qe</groupId>
+        <artifactId>parent</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+        <relativePath>../..</relativePath>
+    </parent>
+    <properties>
+        <!-- Need to be this version, I wasn't able to reproduce QUARKUS-5178 on newer versions-->
+        <quarkus-kafka-streams-processor.version>2.0.1</quarkus-kafka-streams-processor.version>
+    </properties>
+    <artifactId>kafka-processor</artifactId>
+    <name>Quarkus QE TS: Messaging: Reactive Processor Quarkiverse</name>
+    <dependencies>
+        <dependency>
+            <groupId>io.quarkiverse.kafkastreamsprocessor</groupId>
+            <artifactId>quarkus-kafka-streams-processor-api</artifactId>
+            <version>${quarkus-kafka-streams-processor.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkiverse.kafkastreamsprocessor</groupId>
+            <artifactId>quarkus-kafka-streams-processor-impl</artifactId>
+            <version>${quarkus-kafka-streams-processor.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkiverse.kafkastreamsprocessor</groupId>
+            <artifactId>quarkus-kafka-streams-processor</artifactId>
+            <version>${quarkus-kafka-streams-processor.version}</version>
+        </dependency>
+    </dependencies>
+</project>

messaging/kafka-processor/src/main/java/io/quarkus/ts/messaging/kafka/processor/decorator/HeaderDecorator.java

@@ -0,0 +1,37 @@
+package io.quarkus.ts.messaging.kafka.processor.decorator;
+
+import java.nio.charset.StandardCharsets;
+
+import jakarta.annotation.Priority;
+import jakarta.decorator.Decorator;
+import jakarta.decorator.Delegate;
+import jakarta.inject.Inject;
+
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.streams.processor.api.Processor;
+import org.apache.kafka.streams.processor.api.Record;
+
+import io.quarkiverse.kafkastreamsprocessor.api.decorator.processor.ProcessorDecoratorPriorities;
+
+@Decorator
+@Priority(ProcessorDecoratorPriorities.PUNCTUATOR_DECORATION + 2)
+public class HeaderDecorator<KIn, VIn, KOut, VOut> implements Processor<KIn, VIn, KOut, VOut> {
+    private final Processor<KIn, VIn, KOut, VOut> delegate;
+
+    @Inject
+    public HeaderDecorator(@Delegate Processor<KIn, VIn, KOut, VOut> delegate) {
+        this.delegate = delegate;
+    }
+
+    @Override
+    public void process(Record<KIn, VIn> record) {
+        Header header = record.headers().lastHeader("custom-header");
+        if (header != null) {
+            String value = new String(header.value(), StandardCharsets.UTF_8);
+            if (value.contains("error")) {
+                throw new IllegalStateException("Error in header");
+            }
+        }
+        delegate.process(record);
+    }
+}

messaging/kafka-processor/src/main/java/io/quarkus/ts/messaging/kafka/processor/processor/PingProcessor.java

@@ -0,0 +1,15 @@
+package io.quarkus.ts.messaging.kafka.processor.processor;
+
+import org.apache.kafka.streams.processor.api.ContextualProcessor;
+import org.apache.kafka.streams.processor.api.Record;
+
+import io.quarkiverse.kafkastreamsprocessor.api.Processor;
+
+@Processor
+public class PingProcessor extends ContextualProcessor<String, String, String, String> {
+
+    @Override
+    public void process(Record<String, String> ping) {
+        context().forward(ping);
+    }
+}

messaging/kafka-processor/src/main/resources/application.properties

@@ -0,0 +1,3 @@
+kafkastreamsprocessor.input.topic=ping-events
+kafkastreamsprocessor.output.topic=pong-events
+quarkus.kafka-streams.topics=ping-events,pong-events

messaging/kafka-processor/src/test/java/io/quarkus/ts/messaging/kafka/processor/DevModeKafkaProcessorIT.java

@@ -0,0 +1,34 @@
+package io.quarkus.ts.messaging.kafka.processor;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+import io.quarkus.test.bootstrap.RestService;
+import io.quarkus.test.scenarios.QuarkusScenario;
+import io.quarkus.test.services.DevModeQuarkusApplication;
+import io.quarkus.ts.messaging.kafka.processor.decorator.HeaderDecorator;
+
+@Tag("QUARKUS-5178")
+@QuarkusScenario
+public class DevModeKafkaProcessorIT {
+
+    @DevModeQuarkusApplication
+    static RestService app = new RestService().setAutoStart(false);
+
+    /**
+     * The QUARKUS-5178 was caused only in dev mode and not all the time.
+     * The selected 5 runs start stops should be enough to detect the original issue
+     */
+    @Test
+    public void quarkusShouldStartWithoutFailTest() {
+        // As QUARKUS-5178 not occurring all the time so occasionally we need to check the dev mode multiple time
+        for (int i = 0; i < 5; i++) {
+            assertDoesNotThrow(() -> app.start(),
+                    "The QUARKUS-5178 is probably not fixed");
+            app.logs().assertDoesNotContain("java.lang.ClassNotFoundException: " + HeaderDecorator.class.getName());
+            app.stop();
+        }
+    }
+}

pom.xml

@@ -541,6 +541,7 @@
                 <module>env-info</module>
                 <module>messaging/amqp-reactive</module>
                 <module>messaging/qpid</module>
+                <module>messaging/kafka-processor</module>
                 <module>messaging/kafka-streams-reactive-messaging</module>
                 <module>messaging/kafka-confluent-avro-reactive-messaging</module>
                 <module>messaging/kafka-strimzi-avro-reactive-messaging</module>