Add abortStream method to Http1StreamManager and ensure streams are released to connection pool

src/main/java/software/amazon/awssdk/crt/http/Http1StreamManager.java

@@ -7,6 +7,7 @@
 import software.amazon.awssdk.crt.CrtRuntimeException;
 
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * Manages a Pool of HTTP/1.1 Streams. Creates and manages HTTP/1.1 connections
@@ -97,6 +98,11 @@ public CompletableFuture<HttpStream> acquireStream(HttpRequestBase request,
             if (throwable != null) {
                 completionFuture.completeExceptionally(throwable);
             } else {
+                // Guard ensures the connection is released exactly once, regardless of
+                // whether release happens via onResponseComplete, activate failure, or
+                // external cancellation (e.g., SDK timeout calling cancel+close).
+                AtomicBoolean connectionReleased = new AtomicBoolean(false);
+
                 try {
                     HttpStreamBase stream = conn.makeRequest(request, new HttpStreamBaseResponseHandler() {
                         @Override
@@ -118,8 +124,10 @@ public int onResponseBody(HttpStreamBase stream, byte[] bodyBytesIn) {
                         @Override
                         public void onResponseComplete(HttpStreamBase stream, int errorCode) {
                             streamHandler.onResponseComplete(stream, errorCode);
-                            /* Release the connection back */
-                            connManager.releaseConnection(conn);
+                            /* Release the connection back (at most once) */
+                            if (connectionReleased.compareAndSet(false, true)) {
+                                connManager.releaseConnection(conn);
+                            }
                         }
                     }, useManualDataWrites);
                     completionFuture.complete((HttpStream) stream);
@@ -134,11 +142,22 @@ public void onResponseComplete(HttpStreamBase stream, int errorCode) {
                     } catch (CrtRuntimeException e) {
                         /* If activate failed, complete callback will not be invoked */
                         streamHandler.onResponseComplete(stream, e.errorCode);
-                        /* Release the connection back */
+                        if (connectionReleased.compareAndSet(false, true)) {
+                            connManager.releaseConnection(conn);
+                        }
+                    }
+
+                    // If the stream was externally cancelled+closed between
+                    // completionFuture.complete() and stream.activate() above, activate()
+                    // becomes a no-op (stream handle is null) and onResponseComplete will
+                    // never fire. Detect this and release the connection.
+                    if (stream.isNull() && connectionReleased.compareAndSet(false, true)) {
                         connManager.releaseConnection(conn);
                     }
                 } catch (Exception ex) {
-                    connManager.releaseConnection(conn);
+                    if (connectionReleased.compareAndSet(false, true)) {
+                        connManager.releaseConnection(conn);
+                    }
                     completionFuture.completeExceptionally(ex);
                 }
             }

src/main/java/software/amazon/awssdk/crt/http/HttpStreamManager.java

@@ -106,6 +106,31 @@ public int getMaxConnections() {
         }
     }
 
+    /**
+     * Abort an in-flight stream obtained from this manager. Cancels the HTTP exchange
+     * and ensures the underlying connection/stream slot is properly released back to the
+     * pool so it can be reused by new requests.
+     *
+     * <p>This is the correct way to cancel an in-flight request from outside the stream
+     * callback lifecycle (e.g., on SDK timeout). For HTTP/1.1, the connection release is
+     * guaranteed by the {@code AtomicBoolean} guard and {@code isNull()} check inside
+     * {@code Http1StreamManager.acquireStream()} — calling {@code cancel()} triggers
+     * {@code onResponseComplete} on activated streams, or the post-activate guard catches
+     * streams closed before activation. For HTTP/2, the native layer handles stream slot
+     * accounting internally.
+     *
+     * <p>Safe to call with null (e.g., if the stream was never acquired), after normal
+     * completion, or multiple times.
+     *
+     * @param stream the stream to abort, or null
+     */
+    public void abortStream(HttpStreamBase stream) {
+        if (stream != null && !stream.isNull()) {
+            stream.cancel();
+            stream.close();
+        }
+    }
+
     @Override
     public void close() {
         if (this.h1StreamManager != null) {