fix: close getObject response stream for buffering transformers

src/main/java/software/amazon/encryption/s3/S3EncryptionClient.java

@@ -457,13 +457,32 @@ public <T> T getObject(GetObjectRequest getObjectRequest,
                 .commitmentPolicy(_commitmentPolicy)
                 .build();
 
+        ResponseInputStream<GetObjectResponse> joinFutureGet = null;
+        boolean callerOwnsStream = false;
         try {
-            ResponseInputStream<GetObjectResponse> joinFutureGet = pipeline.getObject(getObjectRequest, AsyncResponseTransformer.toBlockingInputStream()).join();
-            return responseTransformer.transform(joinFutureGet.response(), AbortableInputStream.create(joinFutureGet));
+            joinFutureGet = pipeline.getObject(getObjectRequest, AsyncResponseTransformer.toBlockingInputStream()).join();
+            T result = responseTransformer.transform(joinFutureGet.response(), AbortableInputStream.create(joinFutureGet));
+            // Streaming transformers (e.g. toInputStream) hand the stream back to the caller, who is
+            // then responsible for closing it. Buffering transformers (e.g. toBytes, toFile) fully
+            // consume the stream and return a materialized result, so ownership is not transferred and
+            // the stream is closed in the finally block below to release its buffers.
+            callerOwnsStream = responseTransformer.needsConnectionLeftOpen();
+            return result;
         } catch (CompletionException e) {
             throw new S3EncryptionClientException(e.getCause().getMessage(), e.getCause());
         } catch (Exception e) {
             throw new S3EncryptionClientException("Unable to transform response.", e);
+        } finally {
+            // Close the stream unless ownership was successfully handed to the caller. This covers the
+            // buffering case (leak fix) and also the case where transform threw before returning, so a
+            // streaming transformer's stream does not leak when the caller never received it.
+            if (joinFutureGet != null && !callerOwnsStream) {
+                try {
+                    joinFutureGet.close();
+                } catch (IOException e) {
+                    throw new S3EncryptionClientException("Unable to close response stream.", e);
+                }
+            }
         }
     }
 

src/test/java/software/amazon/encryption/s3/S3EncryptionClientGetObjectStreamCloseTest.java

@@ -0,0 +1,36 @@
+// Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.encryption.s3;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.File;
+import java.io.IOException;
+
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.core.sync.ResponseTransformer;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+
+/**
+ * Pins the contract that {@link S3EncryptionClient#getObject} relies on to decide whether to close
+ * the response stream: buffering transformers do not need the connection left open (so the client
+ * closes the stream and avoids the leak), while streaming transformers do (the caller closes it).
+ */
+public class S3EncryptionClientGetObjectStreamCloseTest {
+
+    @Test
+    public void bufferingTransformersDoNotNeedConnectionLeftOpen() throws IOException {
+        assertFalse(ResponseTransformer.<GetObjectResponse>toBytes().needsConnectionLeftOpen());
+
+        File tempFile = File.createTempFile("s3ec-close-test", ".tmp");
+        tempFile.delete(); // toFile requires the file to not already exist
+        tempFile.deleteOnExit();
+        assertFalse(ResponseTransformer.<GetObjectResponse>toFile(tempFile.toPath()).needsConnectionLeftOpen());
+    }
+
+    @Test
+    public void streamingTransformerNeedsConnectionLeftOpen() {
+        assertTrue(ResponseTransformer.<GetObjectResponse>toInputStream().needsConnectionLeftOpen());
+    }
+}