8324209: Check implementation of Expect: 100-continue in the java.net.http.HttpClient

Reviewed-by: dfuchs, jpai

src/java.net.http/share/classes/jdk/internal/net/http/Exchange.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2015, 2023, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2015, 2024, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
@@ -39,6 +39,8 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.function.Function;
 import java.net.http.HttpClient;
 import java.net.http.HttpHeaders;
@@ -453,32 +455,55 @@ final class Exchange<T> {
     // for the 100-Continue response
     private CompletableFuture<Response> expectContinue(ExchangeImpl<T> ex) {
         assert request.expectContinue();
+
+        long responseTimeoutMillis = 5000;
+        if (request.timeout().isPresent()) {
+            final long timeoutMillis = request.timeout().get().toMillis();
+            responseTimeoutMillis = Math.min(responseTimeoutMillis, timeoutMillis);
+        }
+
         return ex.getResponseAsync(parentExecutor)
+                .completeOnTimeout(null, responseTimeoutMillis, TimeUnit.MILLISECONDS)
                 .thenCompose((Response r1) -> {
-            Log.logResponse(r1::toString);
+                    // The response will only be null if there was a timeout
-            int rcode = r1.statusCode();
+                    // send body regardless
-            if (rcode == 100) {
+                    if (r1 == null) {
-                Log.logTrace("Received 100-Continue: sending body");
+                        if (debug.on())
-                if (debug.on()) debug.log("Received 100-Continue for %s", r1);
+                            debug.log("Setting ExpectTimeoutRaised and sending request body");
-                CompletableFuture<Response> cf =
+                        exchImpl.setExpectTimeoutRaised();
-                        exchImpl.sendBodyAsync()
+                        CompletableFuture<Response> cf =
-                                .thenCompose(exIm -> exIm.getResponseAsync(parentExecutor));
+                                exchImpl.sendBodyAsync()
-                cf = wrapForUpgrade(cf);
+                                        .thenCompose(exIm -> exIm.getResponseAsync(parentExecutor));
-                cf = wrapForLog(cf);
+                        cf = wrapForUpgrade(cf);
-                return cf;
+                        cf = wrapForLog(cf);
-            } else {
+                        return cf;
-                Log.logTrace("Expectation failed: Received {0}",
+                    }
-                        rcode);
+
-                if (debug.on()) debug.log("Expect-Continue failed (%d) for: %s", rcode, r1);
+                    Log.logResponse(r1::toString);
-                if (upgrading && rcode == 101) {
+                    int rcode = r1.statusCode();
-                    IOException failed = new IOException(
+                    if (rcode == 100) {
-                            "Unable to handle 101 while waiting for 100");
+                        Log.logTrace("Received 100-Continue: sending body");
-                    return MinimalFuture.failedFuture(failed);
+                        if (debug.on())
-                }
+                            debug.log("Received 100-Continue for %s", r1);
-                exchImpl.expectContinueFailed(rcode);
+                        CompletableFuture<Response> cf =
-                return MinimalFuture.completedFuture(r1);
+                                exchImpl.sendBodyAsync()
-            }
+                                        .thenCompose(exIm -> exIm.getResponseAsync(parentExecutor));
-        });
+                        cf = wrapForUpgrade(cf);
+                        cf = wrapForLog(cf);
+                        return cf;
+                    } else {
+                        Log.logTrace("Expectation failed: Received {0}", rcode);
+                        if (debug.on())
+                            debug.log("Expect-Continue failed (%d) for: %s", rcode, r1);
+                        if (upgrading && rcode == 101) {
+                            IOException failed = new IOException(
+                                    "Unable to handle 101 while waiting for 100");
+                            return MinimalFuture.failedFuture(failed);
+                        }
+                        exchImpl.expectContinueFailed(rcode);
+                        return MinimalFuture.completedFuture(r1);
+                    }
+                });
     }
 
     // After sending the request headers, if no ProxyAuthorizationRequired

src/java.net.http/share/classes/jdk/internal/net/http/ExchangeImpl.java

@@ -58,6 +58,8 @@ abstract class ExchangeImpl<T> {
 
     final Exchange<T> exchange;
 
+    private volatile boolean expectTimeoutRaised;
+
     // this will be set to true only when the peer explicitly states (through a GOAWAY frame or
     // a relevant error code in reset frame) that the corresponding stream (id) wasn't processed
     private volatile boolean unprocessedByPeer;
@@ -71,6 +73,14 @@ abstract class ExchangeImpl<T> {
         return exchange;
     }
 
+    final void setExpectTimeoutRaised() {
+        expectTimeoutRaised = true;
+    }
+
+    final boolean expectTimeoutRaised() {
+        return expectTimeoutRaised;
+    }
+
     HttpClientImpl client() {
         return exchange.client();
     }

src/java.net.http/share/classes/jdk/internal/net/http/Stream.java

@@ -1200,11 +1200,17 @@ class Stream<T> extends ExchangeImpl<T> {
         try {
             if (!response_cfs.isEmpty()) {
                 // This CompletableFuture was created by completeResponse().
-                // it will be already completed.
+                // it will be already completed, unless the expect continue
-                cf = response_cfs.remove(0);
+                // timeout fired
+                cf = response_cfs.get(0);
+                if (cf.isDone()) {
+                    cf = response_cfs.remove(0);
+                }
+
                 // if we find a cf here it should be already completed.
                 // finding a non completed cf should not happen. just assert it.
-                assert cf.isDone() : "Removing uncompleted response: could cause code to hang!";
+                assert cf.isDone() || request.expectContinue && expectTimeoutRaised()
+                        : "Removing uncompleted response: could cause code to hang!";
             } else {
                 // getResponseAsync() is called first. Create a CompletableFuture
                 // that will be completed by completeResponse() when
@@ -1239,7 +1245,7 @@ class Stream<T> extends ExchangeImpl<T> {
             int cfs_len = response_cfs.size();
             for (int i=0; i<cfs_len; i++) {
                 cf = response_cfs.get(i);
-                if (!cf.isDone()) {
+                if (!cf.isDone() && !expectTimeoutRaised()) {
                     Log.logTrace("Completing response (streamid={0}): {1}",
                                  streamid, cf);
                     if (debug.on())
@@ -1247,6 +1253,14 @@ class Stream<T> extends ExchangeImpl<T> {
                     response_cfs.remove(cf);
                     cf.complete(resp);
                     return;
+                } else if (expectTimeoutRaised()) {
+                    Log.logTrace("Completing response (streamid={0}): {1}",
+                            streamid, cf);
+                    if (debug.on())
+                        debug.log("Completing responseCF(%d) with response headers", i);
+                    // The Request will be removed in getResponseAsync()
+                    cf.complete(resp);
+                    return;
                 } // else we found the previous response: just leave it alone.
             }
             cf = MinimalFuture.completedFuture(resp);

test/jdk/java/net/httpclient/ExpectContinueTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022, 2023, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2022, 2024, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
@@ -85,8 +85,8 @@ public class ExpectContinueTest implements HttpServerAdapters {
     Http1HangServer http1HangServer;
     Http2TestServer http2TestServer; // HTTP/2
 
-    URI getUri, postUri, hangUri;
+    URI getUri, postUri, forcePostUri, hangUri;
-    URI h2postUri, h2hangUri, h2endStreamUri, h2warmupURI;
+    URI h2postUri, h2forcePostUri, h2hangUri, h2endStreamUri, h2warmupURI;
 
     static PrintStream err = new PrintStream(System.err);
     static PrintStream out = new PrintStream(System.out);
@@ -97,8 +97,10 @@ public class ExpectContinueTest implements HttpServerAdapters {
         return new Object[][]{
                 // URI, Expected Status Code, Will finish with Exception, Protocol Version
                 { postUri, 200, false, HTTP_1_1 },
+                { forcePostUri, 200, false, HTTP_1_1 },
                 { hangUri, 417, false, HTTP_1_1},
                 { h2postUri, 200, false, HTTP_2 },
+                { h2forcePostUri, 200, false, HTTP_2 },
                 { h2hangUri, 417, false, HTTP_2 },
                 { h2endStreamUri, 200, true, HTTP_2 }, // Error
         };
@@ -127,7 +129,7 @@ public class ExpectContinueTest implements HttpServerAdapters {
             } catch (Exception e) {
                 testThrowable = e.getCause();
             }
-            verifyRequest(expectedStatusCode, resp, exceptionally, testThrowable);
+            verifyRequest(uri.getPath(), expectedStatusCode, resp, exceptionally, testThrowable);
         }
     }
 
@@ -137,8 +139,10 @@ public class ExpectContinueTest implements HttpServerAdapters {
         http1TestServer = HttpTestServer.create(HTTP_1_1);
         http1TestServer.addHandler(new GetHandler(), "/http1/get");
         http1TestServer.addHandler(new PostHandler(), "/http1/post");
+        http1TestServer.addHandler(new ForcePostHandler(), "/http1/forcePost");
         getUri = URI.create("http://" + http1TestServer.serverAuthority() + "/http1/get");
         postUri = URI.create("http://" + http1TestServer.serverAuthority() + "/http1/post");
+        forcePostUri = URI.create("http://" + http1TestServer.serverAuthority() + "/http1/forcePost");
 
         // Due to limitations of the above Http1 Test Server, a manual approach is taken to test the hanging with the
         // httpclient using Http1 so that the correct response header can be returned for the test case
@@ -149,17 +153,19 @@ public class ExpectContinueTest implements HttpServerAdapters {
         http2TestServer.setExchangeSupplier(ExpectContinueTestExchangeImpl::new);
         http2TestServer.addHandler(new GetHandler().toHttp2Handler(), "/http2/warmup");
         http2TestServer.addHandler(new PostHandler().toHttp2Handler(), "/http2/post");
+        http2TestServer.addHandler(new ForcePostHandler().toHttp2Handler(), "/http2/forcePost");
         http2TestServer.addHandler(new PostHandlerCantContinue().toHttp2Handler(), "/http2/hang");
         http2TestServer.addHandler(new PostHandlerHttp2(), "/http2/endStream");
 
         h2warmupURI = new URI("http://" + http2TestServer.serverAuthority() + "/http2/warmup");
         h2postUri = URI.create("http://" + http2TestServer.serverAuthority() + "/http2/post");
+        h2forcePostUri = URI.create("http://" + http2TestServer.serverAuthority() + "/http2/forcePost");
         h2hangUri = URI.create("http://" + http2TestServer.serverAuthority() + "/http2/hang");
         h2endStreamUri = URI.create("http://" + http2TestServer.serverAuthority() + "/http2/endStream");
 
-        out.printf("HTTP/1.1 server listening at: %s", http1TestServer.serverAuthority());
+        out.printf("HTTP/1.1 server listening at: %s %n", http1TestServer.serverAuthority());
-        out.printf("HTTP/1.1 hang server listening at: %s", hangUri.getRawAuthority());
+        out.printf("HTTP/1.1 hang server listening at: %s %n", hangUri.getRawAuthority());
-        out.printf("HTTP/2 clear server listening at: %s", http2TestServer.serverAuthority());
+        out.printf("HTTP/2 clear server listening at: %s %n", http2TestServer.serverAuthority());
 
         http1TestServer.start();
         http1HangServer.start();
@@ -207,6 +213,18 @@ public class ExpectContinueTest implements HttpServerAdapters {
         }
     }
 
+    static class ForcePostHandler implements HttpTestHandler {
+        @Override
+        public void handle(HttpTestExchange exchange) throws IOException {
+            try (InputStream is = exchange.getRequestBody()) {
+                err.println("Server reading body inside the force Post");
+                is.readAllBytes();
+                err.println("Server send 200 (length=0) in the force post");
+                exchange.sendResponseHeaders(200, 0);
+            }
+        }
+    }
+
     static class PostHandlerHttp2 implements Http2Handler {
 
         @Override
@@ -337,15 +355,18 @@ public class ExpectContinueTest implements HttpServerAdapters {
         }
     }
 
-    private void verifyRequest(int expectedStatusCode, HttpResponse<String> resp, boolean exceptionally, Throwable testThrowable) {
+    private void verifyRequest(String path, int expectedStatusCode, HttpResponse<String> resp, boolean exceptionally, Throwable testThrowable) {
+        if (!exceptionally) {
+            err.printf("Response code %s received for path %s %n", resp.statusCode(), path);
+        }
         if (exceptionally && testThrowable != null) {
-            err.println(testThrowable);
+            err.println("Finished exceptionally Test throwable: " + testThrowable);
             assertEquals(IOException.class, testThrowable.getClass());
         } else if (exceptionally) {
             throw new TestException("Expected case to finish with an IOException but testException is null");
         } else if (resp != null) {
             assertEquals(resp.statusCode(), expectedStatusCode);
-            err.println("Request completed successfully");
+            err.println("Request completed successfully for path " + path);
             err.println("Response Headers: " + resp.headers());
             err.println("Response Status Code: " + resp.statusCode());
         }