reproduce exception, README.md

src/main/java/si/puci/GreetingResource.java

@@ -1,16 +1,47 @@
 package si.puci;
 
+import java.time.OffsetDateTime;
+
+import org.eclipse.microprofile.context.ManagedExecutor;
+import org.eclipse.microprofile.context.ThreadContext;
+import org.eclipse.microprofile.reactive.messaging.Channel;
+
+import io.smallrye.context.api.ManagedExecutorConfig;
+import io.smallrye.reactive.messaging.MutinyEmitter;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import jakarta.transaction.Transactional;
 import jakarta.ws.rs.GET;
 import jakarta.ws.rs.Path;
 import jakarta.ws.rs.Produces;
 import jakarta.ws.rs.core.MediaType;
 
 @Path("/hello")
-public class GreetingResource {
+@Transactional
+@ApplicationScoped
+public class GreetingResource
+{
+    @Inject
+    @Channel("words-out")
+    MutinyEmitter<String> emitter;
+
+    @Inject
+    @ManagedExecutorConfig(cleared = ThreadContext.TRANSACTION)
+    ManagedExecutor noTxExecutor;
 
     @GET
     @Produces(MediaType.TEXT_PLAIN)
-    public String hello() {
+    public String hello()
+    {
+        final var myEntity = new MyEntity();
+        myEntity.field = OffsetDateTime.now().toString();
+        myEntity.persistAndFlush();
+
+        emitter.sendAndAwait(myEntity.field);
+
+        //this kinda solves the problem
+        //noTxExecutor.runAsync(() -> emitter.sendAndAwait(myEntity.field)).join();
+
         return "Hello from RESTEasy Reactive";
     }
 }

src/main/resources/application.properties

@@ -1,3 +1,2 @@
-mp.messaging.incoming.words-in.topic=words
 mp.messaging.outgoing.words-out.topic=words
-mp.messaging.incoming.words-in.auto.offset.reset=earliest
+mp.messaging.outgoing.words-out.connector=smallrye-kafka

src/test/java/si/puci/GreetingResourceTest.java

@@ -1,20 +1,57 @@
 package si.puci;
 
+import io.quarkus.logging.Log;
 import io.quarkus.test.junit.QuarkusTest;
+import io.smallrye.mutiny.Multi;
+
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 import static io.restassured.RestAssured.given;
 import static org.hamcrest.CoreMatchers.is;
 
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
 @QuarkusTest
-class GreetingResourceTest {
+class GreetingResourceTest
+{
     @Test
-    void testHelloEndpoint() {
-        given()
-          .when().get("/hello")
-          .then()
-             .statusCode(200)
-             .body(is("Hello from RESTEasy Reactive"));
-    }
+    void testHelloEndpoint()
+    {
+        AtomicBoolean failure = new AtomicBoolean(false);
+        AtomicReference<Throwable> ex = new AtomicReference<>(null);
 
+        final var subscription = Multi.createFrom().ticks().every(Duration.of(1, ChronoUnit.MILLIS))
+                .onItem().invoke(i -> given()
+                        .when().get("/hello")
+                        .then()
+                        .statusCode(200)
+                        .body(is("Hello from RESTEasy Reactive")))
+                .subscribe().with(
+                        Log::info,
+                        err -> {
+                            Log.error(err);
+                            ex.set(err);
+                            failure.set(true);
+                        });
+        try
+        {
+            Thread.sleep(60000);
+        }
+        catch (InterruptedException e)
+        {
+            throw new RuntimeException(e);
+        }
+
+        subscription.cancel();
+        if (ex.get() != null)
+        {
+            Log.error(ex.get());
+        }
+        Assertions.assertFalse(failure.get());
+        //no error!
+    }
 }