8347597: HttpClient: improve exception reporting when closing connection

Reviewed-by: jpai

Author: dfuch

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2015, 2023, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2015, 2025, 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
@@ -72,7 +72,7 @@ public CompletableFuture<Void> finishConnect() {
                     if (ex == null) {
                         return plainConnection.finishConnect();
                     } else {
-                        plainConnection.close();
+                        plainConnection.close(ex);
                         return MinimalFuture.<Void>failedFuture(ex);
                     } })
                 .thenCompose(Function.identity());

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2015, 2023, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2015, 2025, 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
@@ -81,7 +81,7 @@ public CompletableFuture<Void> finishConnect() {
                     if (ex == null) {
                         return plainConnection.finishConnect();
                     } else {
-                        plainConnection.close();
+                        plainConnection.close(ex);
                         return MinimalFuture.<Void>failedFuture(ex);
                     } })
                 .thenCompose(Function.identity());

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2015, 2024, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2015, 2025, 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
@@ -33,11 +33,13 @@
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 import java.net.http.HttpClient;
 import java.net.http.HttpResponse;
 import java.net.http.HttpTimeoutException;
 
+import jdk.internal.net.http.HttpClientImpl.DelegatingExecutor;
 import jdk.internal.net.http.common.Logger;
 import jdk.internal.net.http.common.MinimalFuture;
 import jdk.internal.net.http.common.Utils;
@@ -63,9 +65,9 @@ final class Exchange<T> {
 
     // used to record possible cancellation raised before the exchImpl
     // has been established.
-    private volatile IOException failed;
+    private final AtomicReference<IOException> failed = new AtomicReference<>();
     final MultiExchange<T> multi;
-    final Executor parentExecutor;
+    final DelegatingExecutor parentExecutor;
     volatile boolean upgrading; // to HTTP/2
     volatile boolean upgraded;  // to HTTP/2
     final PushGroup<T> pushGroup;
@@ -91,7 +93,7 @@ PushGroup<T> getPushGroup() {
         return pushGroup;
     }
 
-    Executor executor() {
+    DelegatingExecutor executor() {
         return parentExecutor;
     }
 
@@ -236,26 +238,26 @@ public void cancel(IOException cause) {
         // If the impl is non null, propagate the exception right away.
         // Otherwise record it so that it can be propagated once the
         // exchange impl has been established.
-        ExchangeImpl<?> impl = exchImpl;
+        ExchangeImpl<?> impl;
+        IOException closeReason = null;
+        synchronized (this) {
+            impl =  exchImpl;
+            if (impl == null) {
+                // no impl yet. record the exception
+                failed.compareAndSet(null, cause);
+            }
+        }
         if (impl != null) {
             // propagate the exception to the impl
             if (debug.on()) debug.log("Cancelling exchImpl: %s", exchImpl);
             impl.cancel(cause);
         } else {
-            // no impl yet. record the exception
-            IOException failed = this.failed;
-            if (failed == null) {
-                synchronized (this) {
-                    failed = this.failed;
-                    if (failed == null) {
-                        failed = this.failed = cause;
-                    }
-                }
-            }
-
-            // abort/close the connection if setting up the exchange. This can
+             // abort/close the connection if setting up the exchange. This can
             // be important when setting up HTTP/2
-            connectionAborter.closeConnection(failed);
+            closeReason = failed.get();
+            if (closeReason != null) {
+                connectionAborter.closeConnection(closeReason);
+            }
 
             // now call checkCancelled to recheck the impl.
             // if the failed state is set and the impl is not null, reset
@@ -274,9 +276,9 @@ private void checkCancelled() {
         ExchangeImpl<?> impl = null;
         IOException cause = null;
         CompletableFuture<? extends ExchangeImpl<T>> cf = null;
-        if (failed != null) {
+        if (failed.get() != null) {
             synchronized (this) {
-                cause = failed;
+                cause = failed.get();
                 impl = exchImpl;
                 cf = exchangeCF;
             }
@@ -286,7 +288,11 @@ private void checkCancelled() {
             // The exception is raised by propagating it to the impl.
             if (debug.on()) debug.log("Cancelling exchImpl: %s", impl);
             impl.cancel(cause);
-            failed = null;
+            synchronized (this) {
+                if (impl == exchImpl) {
+                    failed.compareAndSet(cause, null);
+                }
+            }
         } else {
             Log.logTrace("Exchange: request [{0}/timeout={1}ms] no impl is set."
                          + "\n\tCan''t cancel yet with {2}",
@@ -313,7 +319,7 @@ <U> CompletableFuture<U> checkCancelled(CompletableFuture<U> cf, HttpConnection
                         if (t == null) t = new IOException("Request cancelled");
                         if (debug.on()) debug.log("exchange cancelled during connect: " + t);
                         try {
-                            connection.close();
+                            connection.close(t);
                         } catch (Throwable x) {
                             if (debug.on()) debug.log("Failed to close connection", x);
                         }
@@ -330,8 +336,13 @@ public void h2Upgrade() {
         request.setH2Upgrade(this);
     }
 
+    synchronized IOException failed(IOException io) {
+        IOException cause = failed.compareAndExchange(null, io);
+        return cause == null ? io : cause;
+    }
+
     synchronized IOException getCancelCause() {
-        return failed;
+        return failed.get();
     }
 
     // get/set the exchange impl, solving race condition issues with
@@ -409,6 +420,11 @@ private CompletableFuture<Response> checkFor407(ExchangeImpl<T> ex, Throwable t,
         }
     }
 
+    private CompletableFuture<Response> startSendingBody(DelegatingExecutor executor) {
+        return exchImpl.sendBodyAsync()
+                        .thenCompose(exIm -> exIm.getResponseAsync(executor));
+    }
+
     // After sending the request headers, if no ProxyAuthorizationRequired
     // was raised and the expectContinue flag is on, we need to wait
     // for the 100-Continue response
@@ -430,9 +446,7 @@ private CompletableFuture<Response> expectContinue(ExchangeImpl<T> ex) {
                         if (debug.on())
                             debug.log("Setting ExpectTimeoutRaised and sending request body");
                         exchImpl.setExpectTimeoutRaised();
-                        CompletableFuture<Response> cf =
-                                exchImpl.sendBodyAsync()
-                                        .thenCompose(exIm -> exIm.getResponseAsync(parentExecutor));
+                        CompletableFuture<Response> cf = startSendingBody(parentExecutor);
                         cf = wrapForUpgrade(cf);
                         cf = wrapForLog(cf);
                         return cf;
@@ -444,9 +458,7 @@ private CompletableFuture<Response> expectContinue(ExchangeImpl<T> ex) {
                         nonFinalResponses.incrementAndGet();
                         Log.logTrace("Received 100-Continue: sending body");
                         if (debug.on()) debug.log("Received 100-Continue for %s", r1);
-                        CompletableFuture<Response> cf =
-                                exchImpl.sendBodyAsync()
-                                        .thenCompose(exIm -> exIm.getResponseAsync(parentExecutor));
+                        CompletableFuture<Response> cf = startSendingBody(parentExecutor);
                         cf = wrapForUpgrade(cf);
                         cf = wrapForLog(cf);
                         return cf;
@@ -471,8 +483,7 @@ private CompletableFuture<Response> expectContinue(ExchangeImpl<T> ex) {
     private CompletableFuture<Response> sendRequestBody(ExchangeImpl<T> ex) {
         assert !request.expectContinue();
         if (debug.on()) debug.log("sendRequestBody");
-        CompletableFuture<Response> cf = ex.sendBodyAsync()
-                .thenCompose(exIm -> exIm.getResponseAsync(parentExecutor));
+        CompletableFuture<Response> cf = startSendingBody(parentExecutor);
         cf = wrapForUpgrade(cf);
         // after 101 is handled we check for other 1xx responses
         cf = cf.thenCompose(this::ignore1xxResponse);
@@ -669,7 +680,7 @@ HttpResponse.BodySubscriber<T> ignoreBody(HttpResponse.ResponseInfo hdrs) {
                             // Either way, we need to relay it to s.
                             synchronized (this) {
                                 exchImpl = s;
-                                t = failed;
+                                t = failed.get();
                             }
                             // Check whether the HTTP/1.1 was cancelled.
                             if (t == null) t = e.getCancelCause();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2017, 2023, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2017, 2025, 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
@@ -259,8 +259,10 @@ private void flush() {
             checkRequestMore();
 
         } catch (Throwable t) {
-            Throwable x = error;
-            if (x == null) error = t; // will be handled in the finally block
+            synchronized (this) {
+                Throwable x = error;
+                if (x == null) error = t; // will be handled in the finally block
+            }
             if (debug.on()) debug.log("Unexpected error caught in flush()", t);
         } finally {
             // Handles any pending error.
@@ -312,7 +314,7 @@ private void checkForErrors() {
                 // close the upstream connection.
                 Http1Exchange<?> exchg = owner;
                 stop();
-                if (exchg != null) exchg.connection().close();
+                if (exchg != null) exchg.connection().close(x);
             }
         }
     }
@@ -346,7 +348,7 @@ private boolean hasDemand(Http1AsyncDelegate delegate) {
         AbstractSubscription subscription = delegate.subscription();
         long demand = subscription.demand().get();
         if (debug.on())
-            debug.log("downstream subscription demand is %s", demand);
+            debug.log("downstream subscription demand is %s for %s", demand, delegate);
         return demand > 0;
     }
 
@@ -573,7 +575,7 @@ void requestMore() {
             if (canRequestMore.compareAndSet(true, false)) {
                 if (!completed && !dropped) {
                     if (debug.on())
-                        debug.log("Http1TubeSubscriber: requesting one more from upstream");
+                        debug.log("Http1TubeSubscriber: requesting one more from upstream: " + s);
                     s.request(1);
                     return;
                 }