Repository: quarkusio/quarkus
Issue: Incoming HTTP request not being fully read or closed when a constraint validator rejects the request

### Describe the bug

Machine: Apple M1 Pro
JVM: openjdk 11.0.2
Quarkus: 2.7.4.Final, 2.13.3.Final

There seems to be a race condition between reading a request from a TCP socket and constraint validators.

When a request gets rejected by a constraint validator (such as `@Consume`) before Quarkus had a chance to fully ACK the incoming request, Quarkus answers back with the appropriate status code and stops reading the request from the TCP socket but doesn't close the connection.

This behavior, keeping the connection open but not reading data from the socket, leaves clients in a hanging state after they fully filled the TCP transmission window, waiting for it to be ACK'ed.

The client hang may not be observed if the requests are smaller in size, given that in those cases it's likely that the request has been completely received and ACK'ed by Quarkus before the constraint validator had the chance to fire. In these cases, no hanging is observed.

### Expected behavior

Quarkus should either read the incoming request fully or close the connection, so as to signal to the client that the request transmission should be over.

### Actual behavior

Described in the previous section. Follow the steps below to reproduce.

### How to Reproduce?

1. Create a sample Quarkus application using `quarkus create && cd code-with-quarkus`
2. Add the following endpoint:

```java
@Path("/hello")
public class GreetingResource {

    @POST
    @Consumes({"image/jpeg", "image/png"})
    @Produces(MediaType.TEXT_PLAIN)
    public String hello(byte[] imageFile) {
        throw new NotAuthorizedException("");
    }
}
```
3. Make a request with a large payload (3MB is enough on my machine, Apple M1 Pro) which breaks the `@Consumes` filter by sending a different `Content-Type` header. For your convenience, here's a python snippet that'll do it:

```python
import requests
import os

url = "http://localhost:8080/hello"

payload=b"\\x00"+os.urandom(1024 * 1024 * 3)+b"\\x00"
headers = {
  'Content-Type': 'image/gif'
}

response = requests.request("POST", url, headers=headers, data=payload)

print(response.text)
```

At this point, the Python code should hang because the request stream is neither being fully read nor closed by Quarkus.

### Output of `uname -a` or `ver`

(macOS M1 Pro) Darwin <REDACTED>.local 21.6.0 Darwin Kernel Version 21.6.0: Mon Aug 22 20:19:52 PDT 2022; root:xnu-8020.140.49~2/RELEASE_ARM64_T6000 arm64

### Output of `java -version`

openjdk version "11.0.2" 2019-01-15 OpenJDK Runtime Environment 18.9 (build 11.0.2+9) OpenJDK 64-Bit Server VM 18.9 (build 11.0.2+9, mixed mode)

### GraalVM version (if different from Java)

_No response_

### Quarkus version or git rev

Reproduced in 2.7.4.Final and 2.13.3.Final

### Build tool (ie. output of `mvnw --version` or `gradlew --version`)

Apache Maven 3.8.6 Maven home: /Users/<REDACTED>/.m2/wrapper/dists/apache-maven-3.8.6-bin/67568434/apache-maven-3.8.6 Java version: 11.0.2, vendor: Oracle Corporation, runtime: /Library/Java/JavaVirtualMachines/jdk-11.0.2.jdk/Contents/Home Default locale: en_PT, platform encoding: UTF-8 OS name: "mac os x", version: "10.16", arch: "x86_64", family: "mac"

### Additional information

_No response_

Changed files:
extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/HelloResource.java
extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/HelloClient2.java
independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/handlers/ClientSendRequestHandler.java
extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/BasicRestClientTest.java

diff --git a/extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/BasicRestClientTest.java b/extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/BasicRestClientTest.java
index 3148c53acf5..ab885f48c38 100644
--- a/extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/BasicRestClientTest.java
+++ b/extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/BasicRestClientTest.java
@@ -1,7 +1,10 @@
 package io.quarkus.rest.client.reactive;
 
+import static io.restassured.RestAssured.given;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 
+import java.time.Duration;
 import java.util.Map;
 import java.util.Set;
 
@@ -72,4 +75,23 @@ void shouldMapQueryParamsWithSpecialCharacters() {
         assertThat(map.get("p5")).isEqualTo("5");
         assertThat(map.get("p6")).isEqualTo("6");
     }
+
+    /**
+     * Test to reproduce https://github.com/quarkusio/quarkus/issues/28818.
+     */
+    @Test
+    void shouldCloseConnectionsWhenFailures() {
+        // It's using 30 seconds because it's the default timeout to release connections. This timeout should not be taken into
+        // account when there are failures, and we should be able to call 3 times to the service without waiting.
+        await().atMost(Duration.ofSeconds(30))
+                .until(() -> {
+                    for (int call = 0; call < 3; call++) {
+                        given()
+                                .when().get("/hello/callClientForImageInfo")
+                                .then()
+                                .statusCode(500);
+                    }
+                    return true;
+                });
+    }
 }
diff --git a/extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/HelloClient2.java b/extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/HelloClient2.java
index 660c55a0260..2c2c31eab34 100644
--- a/extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/HelloClient2.java
+++ b/extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/HelloClient2.java
@@ -4,6 +4,7 @@
 import javax.ws.rs.GET;
 import javax.ws.rs.POST;
 import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
 import javax.ws.rs.core.MediaType;
 
 import org.eclipse.microprofile.rest.client.inject.RegisterRestClient;
@@ -23,4 +24,10 @@ public interface HelloClient2 {
     @GET
     @Path("delay")
     Uni<String> delay();
+
+    @POST
+    @Path("/imageInfo")
+    @Consumes("image/gif")
+    @Produces(MediaType.TEXT_PLAIN)
+    String imageInfo(byte[] imageFile);
 }
diff --git a/extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/HelloResource.java b/extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/HelloResource.java
index f298c0b3030..65b0dc25760 100644
--- a/extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/HelloResource.java
+++ b/extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/HelloResource.java
@@ -78,4 +78,23 @@ public Uni<String> delay() {
         return Uni.createFrom().item("Hello")
                 .onItem().delayIt().by(Duration.ofMillis(500));
     }
+
+    @Path("callClientForImageInfo")
+    @GET
+    public String callClientForImageInfo() {
+        int size = 1024 * 1024 * 5;
+
+        byte[] buffer = new byte[size];
+
+        //Should provoke 415 Unsupported Media Type
+        return client2.imageInfo(buffer);
+    }
+
+    @POST
+    @Consumes({ "image/jpeg", "image/png" })
+    @Path("/imageInfo")
+    @Produces(MediaType.TEXT_PLAIN)
+    public String imageInfo(byte[] imageFile) {
+        throw new IllegalStateException("This method should never be invoked");
+    }
 }
diff --git a/independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/handlers/ClientSendRequestHandler.java b/independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/handlers/ClientSendRequestHandler.java
index 8d1d7d62c3c..2bf3f434ecb 100644
--- a/independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/handlers/ClientSendRequestHandler.java
+++ b/independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/handlers/ClientSendRequestHandler.java
@@ -20,6 +20,7 @@
 import javax.ws.rs.core.HttpHeaders;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.MultivaluedMap;
+import javax.ws.rs.core.Response;
 import javax.ws.rs.core.Variant;
 
 import org.jboss.logging.Logger;
@@ -243,6 +244,11 @@ public void handle(HttpClientResponse clientResponse) {
                         }
                     }
 
+                    if (Response.Status.Family.familyOf(status) != Response.Status.Family.SUCCESSFUL) {
+                        httpClientRequest.connection().close();
+                        requestContext.resume();
+                    }
+
                     if (isResponseMultipart(requestContext)) {
                         QuarkusMultipartResponseDecoder multipartDecoder = new QuarkusMultipartResponseDecoder(
                                 clientResponse);
@@ -366,17 +372,16 @@ public void handle(Buffer buffer) {
                     requestContext.resume(t);
                 }
             }
-        })
-                .onFailure(new Handler<>() {
-                    @Override
-                    public void handle(Throwable failure) {
-                        if (failure instanceof IOException) {
-                            requestContext.resume(new ProcessingException(failure));
-                        } else {
-                            requestContext.resume(failure);
-                        }
-                    }
-                });
+        }).onFailure(new Handler<>() {
+            @Override
+            public void handle(Throwable failure) {
+                if (failure instanceof IOException) {
+                    requestContext.resume(new ProcessingException(failure));
+                } else {
+                    requestContext.resume(failure);
+                }
+            }
+        });
     }
 
     private boolean isResponseMultipart(RestClientRequestContext requestContext) {