Merge pull request #41133 from gsmet/3.11.2-backports-2

[3.11] 3.11.2 backports 2

bom/application/pom.xml

@@ -58,7 +58,7 @@
         <smallrye-graphql.version>2.8.4</smallrye-graphql.version>
         <smallrye-fault-tolerance.version>6.3.0</smallrye-fault-tolerance.version>
         <smallrye-jwt.version>4.5.2</smallrye-jwt.version>
-        <smallrye-context-propagation.version>2.1.0</smallrye-context-propagation.version>
+        <smallrye-context-propagation.version>2.1.2</smallrye-context-propagation.version>
         <smallrye-reactive-streams-operators.version>1.0.13</smallrye-reactive-streams-operators.version>
         <smallrye-reactive-types-converter.version>3.0.1</smallrye-reactive-types-converter.version>
         <smallrye-mutiny-vertx-binding.version>3.12.0</smallrye-mutiny-vertx-binding.version>

core/runtime/src/main/java/io/quarkus/runtime/shutdown/ShutdownRecorder.java

@@ -28,6 +28,9 @@ public void setListeners(List<ShutdownListener> listeners, boolean delayEnabled)
     }
 
     public static void runShutdown() {
+        if (shutdownListeners == null) { // when QUARKUS_INIT_AND_EXIT is used, ShutdownRecorder#setListeners has not been called
+            return;
+        }
         log.debug("Attempting to gracefully shutdown.");
         try {
             executePreShutdown();

docs/src/main/asciidoc/gradle-tooling.adoc

@@ -294,7 +294,7 @@ Before you start Quarkus on the remote host set the environment variable `QUARKU
 on bare metal you can set it via the `export QUARKUS_LAUNCH_DEVMODE=true` command and then run the application with the proper `java -jar ...` command to run the application.
 
 If you plan on running the application via Docker, then you'll need to add `-e QUARKUS_LAUNCH_DEVMODE=true` to the `docker run` command.
-When the application starts you should now see the following line in the logs: `Profile dev activated. Live Coding activated`.
+When the application starts you should now see the following line in the logs: `Profile dev activated. Live Coding activated`. You will also need to give the application the rights to update the deployment resources by adding `RUN chmod o+rw -R /deployments` after the `COPY` commands into your Dockerfile. For security reasons, this option should not be added to the production Dockerfile.
 
 
 NOTE: The remote side does not need to include Maven or any other development tools. The normal `fast-jar` Dockerfile

docs/src/main/asciidoc/maven-tooling.adoc

@@ -238,8 +238,8 @@ include::{includes}/devtools/build.adoc[]
 Before you start Quarkus on the remote host set the environment variable `QUARKUS_LAUNCH_DEVMODE=true`. If you are
 on bare metal you can set it via the `export QUARKUS_LAUNCH_DEVMODE=true` command and then run the application with the proper `java -jar ...` command to run the application.
 
-If you plan on running the application via Docker, then you'll need to add `-e QUARKUS_LAUNCH_DEVMODE=true` to the `docker run` command.
-When the application starts you should now see the following line in the logs: `Profile dev activated. Live Coding activated`.
+If you plan on running the application via Docker, then you'll need to add `-e QUARKUS_LAUNCH_DEVMODE=true` to the `docker run` command. 
+When the application starts you should now see the following line in the logs: `Profile dev activated. Live Coding activated`. You will also need to give the application the rights to update the deployment resources by adding `RUN chmod o+rw -R /deployments` after the `COPY` commands into your Dockerfile. For security reasons, this option should not be added to the production Dockerfile.
 
 NOTE: The remote side does not need to include Maven or any other development tools. The normal `fast-jar` Dockerfile
 that is generated with a new Quarkus application is all you need. If you are using bare metal launch the Quarkus runner

extensions/hibernate-orm/runtime/src/main/java/io/quarkus/hibernate/orm/runtime/session/TransactionScopedStatelessSession.java

@@ -360,7 +360,7 @@ public void refresh(String s, Object o, LockMode lockMode) {
     public void fetch(Object o) {
         checkBlocking();
         try (SessionResult emr = acquireSession()) {
-            emr.statelessSession.refresh(o);
+            emr.statelessSession.fetch(o);
         }
     }
 

extensions/vertx/deployment/src/test/java/io/quarkus/vertx/deployment/MessageConsumerFailureTest.java

@@ -46,14 +46,14 @@ public void testFailure() throws InterruptedException {
         verifyFailure("foo-completion-stage", "java.lang.NullPointerException: Something is null", false);
         verifyFailure("foo-completion-stage-failure", "boom", true);
         verifyFailure("foo-uni", "java.lang.NullPointerException: Something is null", false);
-        verifyFailure("foo-uni-failure", "boom", true);
+        verifyFailure("foo-uni-failure", "java.io.IOException: boom", true);
 
         verifyFailure("foo-blocking", "java.lang.IllegalStateException: Red is dead", false);
         verifyFailure("foo-message-blocking", "java.lang.NullPointerException", false);
         verifyFailure("foo-completion-stage-blocking", "java.lang.NullPointerException: Something is null", false);
         verifyFailure("foo-completion-stage-failure-blocking", "boom", true);
         verifyFailure("foo-uni-blocking", "java.lang.NullPointerException: Something is null", false);
-        verifyFailure("foo-uni-failure-blocking", "boom", true);
+        verifyFailure("foo-uni-failure-blocking", "java.io.IOException: boom", true);
     }
 
     @Test

extensions/websockets-next/deployment/src/test/java/io/quarkus/websockets/next/test/errors/MultiClosedConnectionTest.java

@@ -0,0 +1,96 @@
+package io.quarkus.websockets.next.test.errors;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.net.URI;
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import jakarta.inject.Inject;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.logging.Log;
+import io.quarkus.test.QuarkusUnitTest;
+import io.quarkus.test.common.http.TestHTTPResource;
+import io.quarkus.websockets.next.OnError;
+import io.quarkus.websockets.next.OnOpen;
+import io.quarkus.websockets.next.WebSocket;
+import io.quarkus.websockets.next.WebSocketConnection;
+import io.quarkus.websockets.next.test.utils.WSClient;
+import io.smallrye.mutiny.Multi;
+import io.vertx.core.Vertx;
+import io.vertx.core.impl.NoStackTraceThrowable;
+
+public class MultiClosedConnectionTest {
+
+    @RegisterExtension
+    public static final QuarkusUnitTest test = new QuarkusUnitTest()
+            .withApplicationRoot(root -> {
+                root.addClasses(Echo.class, WSClient.class);
+            });
+
+    @Inject
+    Vertx vertx;
+
+    @TestHTTPResource("echo")
+    URI testUri;
+
+    @Test
+    void testError() throws InterruptedException {
+        WSClient client = WSClient.create(vertx).connect(testUri);
+        client.waitForMessages(1);
+        client.close();
+        assertTrue(Echo.TERMINATION_LATCH.await(5, TimeUnit.SECONDS));
+        assertTrue(Echo.ERROR_LATCH.await(5, TimeUnit.SECONDS));
+        // Connection is closed and the returned Multi should be cancelled
+        int numOfMessages = Echo.MESSAGES.size();
+        Thread.sleep(600);
+        // No more ticks are emitted
+        assertEquals(numOfMessages, Echo.MESSAGES.size());
+    }
+
+    @WebSocket(path = "/echo")
+    public static class Echo {
+
+        static final CountDownLatch TERMINATION_LATCH = new CountDownLatch(1);
+        static final CountDownLatch ERROR_LATCH = new CountDownLatch(1);
+
+        static final List<String> MESSAGES = new CopyOnWriteArrayList<>();
+
+        @OnOpen
+        Multi<String> onOpen() {
+            return Multi.createFrom()
+                    .ticks()
+                    .every(Duration.ofMillis(300))
+                    .map(tick -> tick + "")
+                    .invoke(s -> {
+                        Log.infof("Next tick: %s", s);
+                        MESSAGES.add(s);
+                    })
+                    .onTermination()
+                    .invoke(() -> {
+                        Log.info("Terminated!");
+                        TERMINATION_LATCH.countDown();
+                    });
+        }
+
+        @OnError
+        void onConnectionClosedError(NoStackTraceThrowable t, WebSocketConnection conn) {
+            Log.info("Error callback!");
+            if (conn.isClosed()) {
+                String message = t.getMessage();
+                if (message != null && message.contains("WebSocket is closed")) {
+                    ERROR_LATCH.countDown();
+                }
+            }
+        }
+
+    }
+
+}

extensions/websockets-next/runtime/src/main/java/io/quarkus/websockets/next/runtime/WebSocketEndpointBase.java

@@ -319,41 +319,31 @@ public Uni<Void> sendText(String message, boolean broadcast) {
         return broadcast ? connection.broadcast().sendText(message) : connection.sendText(message);
     }
 
-    public Uni<Void> multiText(Multi<Object> multi, Function<Object, Uni<Void>> action) {
-        multi.onFailure().recoverWithMulti(t -> doOnError(t).toMulti())
+    public Uni<Void> multiText(Multi<Object> multi, Function<? super Object, Uni<?>> action) {
+        multi
+                // Encode and send message
+                .onItem().call(action)
+                .onFailure().recoverWithMulti(t -> {
+                    return doOnError(t).toMulti();
+                })
                 .subscribe().with(
-                        m -> {
-                            // Encode and send message
-                            action.apply(m)
-                                    .onFailure().recoverWithUni(this::doOnError)
-                                    .subscribe()
-                                    .with(v -> LOG.debugf("Multi >> text message: %s", connection),
-                                            t -> LOG.errorf(t, "Unable to send text message from Multi: %s", connection));
-                        },
-                        t -> {
-                            LOG.errorf(t, "Unable to send text message from Multi: %s ", connection);
-                        });
+                        m -> LOG.debugf("Multi >> text message: %s", connection),
+                        t -> LOG.errorf(t, "Unable to send text message from Multi: %s ", connection));
         return Uni.createFrom().voidItem();
     }
 
     public Uni<Void> sendBinary(Buffer message, boolean broadcast) {
         return broadcast ? connection.broadcast().sendBinary(message) : connection.sendBinary(message);
     }
 
-    public Uni<Void> multiBinary(Multi<Object> multi, Function<Object, Uni<Void>> action) {
-        multi.onFailure().recoverWithMulti(t -> doOnError(t).toMulti())
+    public Uni<Void> multiBinary(Multi<Object> multi, Function<? super Object, Uni<?>> action) {
+        multi
+                // Encode and send message
+                .onItem().call(action)
+                .onFailure().recoverWithMulti(t -> doOnError(t).toMulti())
                 .subscribe().with(
-                        m -> {
-                            // Encode and send message
-                            action.apply(m)
-                                    .onFailure().recoverWithUni(this::doOnError)
-                                    .subscribe()
-                                    .with(v -> LOG.debugf("Multi >> binary message: %s", connection),
-                                            t -> LOG.errorf(t, "Unable to send binary message from Multi: %s", connection));
-                        },
-                        t -> {
-                            LOG.errorf(t, "Unable to send text message from Multi: %s ", connection);
-                        });
+                        m -> LOG.debugf("Multi >> binary message: %s", connection),
+                        t -> LOG.errorf(t, "Unable to send binary message from Multi: %s ", connection));
         return Uni.createFrom().voidItem();
     }
 }

independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/core/serialization/DynamicEntityWriter.java

@@ -41,8 +41,15 @@ public void write(ResteasyReactiveRequestContext context, Object entity) throws
             ServerHttpRequest vertxRequest = context.serverRequest();
             // first check and see if the resource method defined a media type and try to use it
             if ((context.getTarget() != null) && (context.getTarget().getProduces() != null)) {
-                MediaType negotiatedMediaType = context.getTarget().getProduces()
-                        .negotiateProduces(vertxRequest.getRequestHeader(HttpHeaders.ACCEPT)).getKey();
+                MediaType negotiatedMediaType = null;
+                List<String> accepts = context.getHttpHeaders().getRequestHeader(HttpHeaders.ACCEPT);
+                for (String accept : accepts) {
+                    negotiatedMediaType = context.getTarget().getProduces().negotiateProduces(accept).getKey();
+                    if (negotiatedMediaType != null) {
+                        break;
+                    }
+                }
+
                 List<MessageBodyWriter<?>> writersList = serialisers.findWriters(null, entity.getClass(), negotiatedMediaType,
                         RuntimeType.SERVER);
                 if (!writersList.isEmpty()) {

independent-projects/resteasy-reactive/server/vertx/src/test/java/org/jboss/resteasy/reactive/server/vertx/test/matching/PreMatchAcceptInHeaderTest.java

@@ -20,6 +20,7 @@
 import jakarta.ws.rs.core.HttpHeaders;
 import jakarta.ws.rs.core.MediaType;
 import jakarta.ws.rs.core.MultivaluedMap;
+import jakarta.ws.rs.core.Response;
 import jakarta.ws.rs.ext.Provider;
 
 import org.jboss.resteasy.reactive.server.spi.ResteasyReactiveResourceInfo;
@@ -123,6 +124,37 @@ void entityTextWithAcceptToTextInFilter() {
                 .body(equalTo("text"));
     }
 
+    @Test
+    void responseEntityJsonWithoutAcceptToTextInFilter() {
+        given().accept("application/json")
+                .when()
+                .get("test/response")
+                .then()
+                .statusCode(200)
+                .body(containsString("\"text\""));
+    }
+
+    @Test
+    void responseEntityTextWithoutAcceptToTextInFilter() {
+        given().accept("text/plain")
+                .when()
+                .get("test/response")
+                .then()
+                .statusCode(200)
+                .body(equalTo("text"));
+    }
+
+    @Test
+    void responseEntityTextWithAcceptToTextInFilter() {
+        given().accept("application/json")
+                .header("x-set-accept-to-text", "true")
+                .when()
+                .get("test/response")
+                .then()
+                .statusCode(200)
+                .body(equalTo("text"));
+    }
+
     @Path("/test")
     public static class Resource {
 
@@ -152,6 +184,13 @@ public String html() {
         public Entity entity() {
             return new Entity("text");
         }
+
+        @GET
+        @Path("response")
+        @Produces({ MediaType.TEXT_PLAIN, MediaType.APPLICATION_JSON })
+        public Response response() {
+            return Response.ok(new Entity("text")).build();
+        }
     }
 
     public record Entity(String value) {

integration-tests/smallrye-context-propagation/src/test/java/io/quarkus/context/test/mutiny/MutinyContextEndpoint.java

@@ -1,6 +1,7 @@
 package io.quarkus.context.test.mutiny;
 
 import java.net.MalformedURLException;
+import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -461,4 +462,21 @@ public Flow.Publisher<String> transactionPropagationWithMulti2() {
         Assertions.assertEquals(2, Person.deleteAll());
         return ret;
     }
+
+    @GET
+    @Path("/bug40852")
+    public String bug40852() {
+        var futureW = Uni
+                .createFrom()
+                .item("item")
+                .onItem()
+                .delayIt()
+                .by(Duration.ofMillis(100))
+                .subscribeAsCompletionStage();
+
+        futureW.whenComplete((result, error) -> {
+            Assertions.assertEquals(true, futureW.isDone());
+        }).join();
+        return "OK";
+    }
 }

integration-tests/smallrye-context-propagation/src/test/java/io/quarkus/context/test/mutiny/MutinyContextPropagationTest.java

@@ -156,6 +156,12 @@ public void testTransactionContextPropagationMulti() {
                 .statusCode(Response.Status.OK.getStatusCode()));
     }
 
+    @Test
+    public void testContextPropagationBug40852() {
+        RestAssured.when().get("/mutiny-context/bug40852").then()
+                .statusCode(Response.Status.OK.getStatusCode());
+    }
+
     private void awaitState(ThrowingRunnable task) {
         Awaitility.await().atMost(5, TimeUnit.SECONDS)
                 .pollInterval(100, TimeUnit.MILLISECONDS)