diff --git a/src/main/java/software/amazon/encryption/s3/S3EncryptionClient.java b/src/main/java/software/amazon/encryption/s3/S3EncryptionClient.java index 3138b2b60..d526a2567 100644 --- a/src/main/java/software/amazon/encryption/s3/S3EncryptionClient.java +++ b/src/main/java/software/amazon/encryption/s3/S3EncryptionClient.java @@ -457,13 +457,32 @@ public T getObject(GetObjectRequest getObjectRequest, .commitmentPolicy(_commitmentPolicy) .build(); + ResponseInputStream joinFutureGet = null; + boolean callerOwnsStream = false; try { - ResponseInputStream 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); + } + } } } diff --git a/src/test/java/software/amazon/encryption/s3/S3EncryptionClientGetObjectStreamCloseTest.java b/src/test/java/software/amazon/encryption/s3/S3EncryptionClientGetObjectStreamCloseTest.java new file mode 100644 index 000000000..959f2f529 --- /dev/null +++ b/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.toBytes().needsConnectionLeftOpen()); + + File tempFile = File.createTempFile("s3ec-close-test", ".tmp"); + tempFile.delete(); // toFile requires the file to not already exist + tempFile.deleteOnExit(); + assertFalse(ResponseTransformer.toFile(tempFile.toPath()).needsConnectionLeftOpen()); + } + + @Test + public void streamingTransformerNeedsConnectionLeftOpen() { + assertTrue(ResponseTransformer.toInputStream().needsConnectionLeftOpen()); + } +}