Allow calling close after abort on a ResponseInputStream.

Before this change, calling `close` after `abort` would raise an IOException. It's well-understood that streams should be `close`d after they're done being used. This was not possible before this change, if `abort` was called.

As a side effect of this change, `abort` now calls `close` on the underlying input stream. This is the behavior of the 1.x SDK, so it is low-risk.

Author: millems

.changes/next-release/bugfix-AWSSDKforJavav2-63a492c.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS SDK for Java v2",
+    "contributor": "",
+    "description": "Fixed an issue where invoking `abort` and then `close` on a `ResponseInputStream` would cause the `close` to fail."
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/ResponseInputStream.java

@@ -20,6 +20,7 @@
 import software.amazon.awssdk.core.io.SdkFilterInputStream;
 import software.amazon.awssdk.http.Abortable;
 import software.amazon.awssdk.http.AbortableInputStream;
+import software.amazon.awssdk.utils.IoUtils;
 import software.amazon.awssdk.utils.Validate;
 
 /**
@@ -57,10 +58,15 @@ public ResponseT response() {
         return response;
     }
 
+    /**
+     * Close the underlying connection, dropping all remaining data in the stream, and not leaving the
+     * connection open to be used for future requests.
+     */
     @Override
     public void abort() {
         if (abortable != null) {
             abortable.abort();
         }
+        IoUtils.closeQuietly(in, null);
     }
 }

core/sdk-core/src/test/java/software/amazon/awssdk/core/ResponseInputStreamTest.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.core;
+
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.mockito.Mockito.never;
+
+import java.io.IOException;
+import java.io.InputStream;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import software.amazon.awssdk.http.Abortable;
+import software.amazon.awssdk.http.AbortableInputStream;
+
+class ResponseInputStreamTest {
+    @Test
+    void abort_withAbortable_closesUnderlyingStream() throws IOException {
+        InputStream stream = Mockito.mock(InputStream.class);
+        Abortable abortable = Mockito.mock(Abortable.class);
+        AbortableInputStream abortableInputStream = AbortableInputStream.create(stream, abortable);
+        ResponseInputStream<Object> responseInputStream = new ResponseInputStream<>(new Object(), abortableInputStream);
+
+        responseInputStream.abort();
+
+        Mockito.verify(abortable).abort();
+        Mockito.verify(stream).close();
+    }
+
+    @Test
+    void failedClose_withinAbort_isIgnored() throws IOException {
+        InputStream stream = Mockito.mock(InputStream.class);
+        Abortable abortable = Mockito.mock(Abortable.class);
+        AbortableInputStream abortableInputStream = AbortableInputStream.create(stream, abortable);
+        ResponseInputStream<Object> responseInputStream = new ResponseInputStream<>(new Object(), abortableInputStream);
+
+        Mockito.doThrow(new IOException()).when(stream).close();
+        assertThatCode(responseInputStream::abort).doesNotThrowAnyException();
+
+        Mockito.verify(abortable).abort();
+        Mockito.verify(stream).close();
+    }
+
+    @Test
+    void abort_withoutAbortable_closesUnderlyingStream() throws IOException {
+        InputStream stream = Mockito.mock(InputStream.class);
+        ResponseInputStream<Object> responseInputStream = new ResponseInputStream<>(new Object(), stream);
+
+        responseInputStream.abort();
+
+        Mockito.verify(stream).close();
+    }
+
+    @Test
+    void close_withAbortable_closesUnderlyingStream() throws IOException {
+        InputStream stream = Mockito.mock(InputStream.class);
+        Abortable abortable = Mockito.mock(Abortable.class);
+        AbortableInputStream abortableInputStream = AbortableInputStream.create(stream, abortable);
+        ResponseInputStream<Object> responseInputStream = new ResponseInputStream<>(new Object(), abortableInputStream);
+
+        responseInputStream.close();
+
+        Mockito.verify(abortable, never()).abort();
+        Mockito.verify(stream).close();
+    }
+}

services/s3/src/it/java/software/amazon/awssdk/services/s3/GetObjectIntegrationTest.java

@@ -16,6 +16,7 @@
 package software.amazon.awssdk.services.s3;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.junit.Assert.assertEquals;
 import static software.amazon.awssdk.testutils.service.S3BucketUtils.temporaryBucketName;
 
@@ -123,6 +124,13 @@ public void contentRangeIsReturnedForRangeRequests() {
         assertThat(stream.response().contentRange()).isEqualTo("bytes 0-1/10000");
     }
 
+    @Test
+    public void sync_closeAfterAbort_doesNotThrowException() {
+        ResponseInputStream<GetObjectResponse> stream = s3.getObject(getObjectRequest);
+        stream.abort();
+        assertThatCode(stream::close).doesNotThrowAnyException();
+    }
+
     private S3Client createClientWithInterceptor(ExecutionInterceptor interceptor) {
         return s3ClientBuilder().overrideConfiguration(ClientOverrideConfiguration.builder()
                                                                                   .addExecutionInterceptor(interceptor)