diff --git a/.changes/next-release/bugfix-AWSSDKforJavav2-fd526c8.json b/.changes/next-release/bugfix-AWSSDKforJavav2-fd526c8.json new file mode 100644 index 000000000000..fa03597dd2ea --- /dev/null +++ b/.changes/next-release/bugfix-AWSSDKforJavav2-fd526c8.json @@ -0,0 +1,6 @@ +{ + "type": "bugfix", + "category": "AWS SDK for Java v2", + "contributor": "", + "description": "Update `ResponseInputStream`, to only return `0` from `available()` if the stream is closed; in other cases, if the wrapped stream returns 0 from `available()`, `1` is returned instead. This is to avoid cases where other classes such as [`java.util.zip.GZIPInputStream`](https://docs.oracle.com/javase/8/docs/api/java/util/zip/GZIPInputStream.html) misinterpret this as meaning the stream is closed." +} diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/ResponseInputStream.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/ResponseInputStream.java index 8f87186d0edd..8d121772c809 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/ResponseInputStream.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/ResponseInputStream.java @@ -118,6 +118,27 @@ public int read(byte[] b, int off, int len) throws IOException { return super.read(b, off, len); } + /** + * Note: Unlike the default implementation, this class will only return 0 if the stream is closed. This is to avoid cases + * where other classes mistake this value as a signal that the stream is closed when it isn't. + *

+ * {@inheritDoc} + */ + @Override + public int available() throws IOException { + if (isClosed()) { + return 0; + } + + int estimate = super.available(); + // Some utilities like java.util.zip.GZIPInputStream use this incorrectly to determine if the stream is *closed* when the + // return value is 0. Guard against misuse of this by returning 1 instead of 0. + // See: + // - https://bugs.openjdk.org/browse/JDK-7036144 + // - https://bugs.openjdk.org/browse/JDK-8374644 + return estimate == 0 ? 1 : estimate; + } + private void cancelTimeoutTask() { if (!hasRead && timeoutTask != null) { timeoutTask.cancel(false); diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/io/SdkFilterInputStream.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/io/SdkFilterInputStream.java index 6016af712678..e5d6eb8c3144 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/io/SdkFilterInputStream.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/io/SdkFilterInputStream.java @@ -18,6 +18,7 @@ import java.io.FilterInputStream; import java.io.IOException; import java.io.InputStream; +import java.util.concurrent.atomic.AtomicBoolean; import software.amazon.awssdk.annotations.SdkProtectedApi; import software.amazon.awssdk.core.exception.AbortedException; import software.amazon.awssdk.core.internal.io.Releasable; @@ -28,6 +29,7 @@ */ @SdkProtectedApi public class SdkFilterInputStream extends FilterInputStream implements Releasable { + private final AtomicBoolean closed = new AtomicBoolean(false); protected SdkFilterInputStream(InputStream in) { super(in); @@ -41,6 +43,7 @@ protected SdkFilterInputStream(InputStream in) { */ protected final void abortIfNeeded() { if (Thread.currentThread().isInterrupted()) { + closed.compareAndSet(false, true); abort(); // execute subclass specific abortion logic throw AbortedException.builder().build(); } @@ -54,16 +57,24 @@ protected void abort() { // no-op by default, but subclass such as S3ObjectInputStream may override } + protected boolean isClosed() { + return closed.get(); + } + @Override public int read() throws IOException { abortIfNeeded(); - return in.read(); + int b = in.read(); + closed.compareAndSet(false, b == -1); + return b; } @Override public int read(byte[] b, int off, int len) throws IOException { abortIfNeeded(); - return in.read(b, off, len); + int nRead = in.read(b, off, len); + closed.compareAndSet(false, nRead == -1); + return nRead; } @Override @@ -80,6 +91,7 @@ public int available() throws IOException { @Override public void close() throws IOException { + closed.compareAndSet(false, true); in.close(); abortIfNeeded(); } @@ -94,6 +106,7 @@ public synchronized void mark(int readlimit) { public synchronized void reset() throws IOException { abortIfNeeded(); in.reset(); + closed.set(false); } @Override @@ -104,6 +117,7 @@ public boolean markSupported() { @Override public void release() { + closed.compareAndSet(false, true); // Don't call IOUtils.release(in, null) or else could lead to infinite loop IoUtils.closeQuietly(this, null); if (in instanceof Releasable) { diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/ResponseInputStreamTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/ResponseInputStreamTest.java index 6710465a899d..4dd2da24be2b 100644 --- a/core/sdk-core/src/test/java/software/amazon/awssdk/core/ResponseInputStreamTest.java +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/ResponseInputStreamTest.java @@ -19,6 +19,7 @@ import static org.assertj.core.api.Assertions.assertThatCode; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import java.io.IOException; import java.io.InputStream; @@ -102,7 +103,7 @@ void customTimeout_noRead_abortsAfterTimeout() throws Exception { } @Test - void customTimouet_readBeforeTimeout_cancelsTimeout() throws Exception { + void customTimeout_readBeforeTimeout_cancelsTimeout() throws Exception { ResponseInputStream responseInputStream = responseInputStream(Duration.ofSeconds(1)); responseInputStream.read(); Thread.sleep(2000); @@ -133,6 +134,27 @@ void negativeTimeout_disablesTimeout() throws Exception { assertThat(responseInputStream.hasTimeoutTask()).isFalse(); } + @Test + void zeroAvailable_availableReturns1() throws IOException { + when(stream.available()).thenReturn(0); + ResponseInputStream responseInputStream = responseInputStream(Duration.ZERO); + assertThat(responseInputStream.available()).isEqualTo(1); + } + + @Test + void nonZeroAvailable_availableReturnsValue() throws IOException { + when(stream.available()).thenReturn(42); + ResponseInputStream responseInputStream = responseInputStream(Duration.ZERO); + assertThat(responseInputStream.available()).isEqualTo(42); + } + + @Test + void closed_AvailableReturns0() throws IOException { + ResponseInputStream responseInputStream = responseInputStream(Duration.ZERO); + responseInputStream.close(); + assertThat(responseInputStream.available()).isZero(); + } + private ResponseInputStream responseInputStream(Duration timeout) { return new ResponseInputStream<>(new Object(), abortableInputStream, timeout); } diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/http/ContentStreamProviderWireMockTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/http/ContentStreamProviderWireMockTest.java index c544254dfb38..67556f65dd37 100644 --- a/core/sdk-core/src/test/java/software/amazon/awssdk/core/http/ContentStreamProviderWireMockTest.java +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/http/ContentStreamProviderWireMockTest.java @@ -122,7 +122,7 @@ public void close() throws IOException { isClosed = true; } - boolean isClosed() { + public boolean isClosed() { return isClosed; } } diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/io/SdkFilterInputStreamTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/io/SdkFilterInputStreamTest.java new file mode 100644 index 000000000000..1280202cc0ac --- /dev/null +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/io/SdkFilterInputStreamTest.java @@ -0,0 +1,138 @@ +/* + * 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.io; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.io.InputStream; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.core.exception.AbortedException; + +public class SdkFilterInputStreamTest { + private InputStream mockStream; + private TestInputStream filterInputStream; + + @BeforeEach + void setup() { + mockStream = mock(InputStream.class); + filterInputStream = new TestInputStream(mockStream); + } + + @Test + void isClosed_streamClosed_returnsTrue() throws IOException { + filterInputStream.close(); + assertThat(filterInputStream.isClosed()).isTrue(); + } + + @Test + void isClosed_readByteReturnsMinus1_returnsTrue() throws IOException { + when(mockStream.read()).thenReturn(-1); + filterInputStream.read(); + assertThat(filterInputStream.isClosed()).isTrue(); + } + + // Possible if two threads try to read the stream at the same time. If one of them sees -1, that value should be "sticky". + @Test + void isClosed_readByte_returnsMinus1ThenPos_returnsTrue() throws IOException { + when(mockStream.read()).thenReturn(-1); + filterInputStream.read(); + assertThat(filterInputStream.isClosed()).isTrue(); + + when(mockStream.read()).thenReturn(42); + filterInputStream.read(); + assertThat(filterInputStream.isClosed()).isTrue(); + } + + @Test + void isClosed_readByteArray_returnsMinus1ThenPos_returnsTrue() throws IOException { + when(mockStream.read(any(byte[].class), anyInt(), anyInt())).thenReturn(-1); + filterInputStream.read(new byte[0], 0, 0); + assertThat(filterInputStream.isClosed()).isTrue(); + + when(mockStream.read(any(byte[].class), anyInt(), anyInt())).thenReturn(42); + filterInputStream.read(new byte[0], 0, 0); + assertThat(filterInputStream.isClosed()).isTrue(); + } + + + @Test + void isClosed_readByteBytesReturnsMinus1_returnsTrue() throws IOException { + when(mockStream.read(any(byte[].class), anyInt(), anyInt())).thenReturn(-1); + filterInputStream.read(new byte[0], 0, 0); + assertThat(filterInputStream.isClosed()).isTrue(); + } + + @Test + void isClosed_released_returnsTrue() { + filterInputStream.release(); + assertThat(filterInputStream.isClosed()).isTrue(); + } + + @Test + void isClosed_abortsDueToInterrupt_returnsTrue() { + Thread.currentThread().interrupt(); + try { + assertThatThrownBy(filterInputStream::read).isInstanceOf(AbortedException.class); + assertThat(filterInputStream.isClosed()).isTrue(); + } finally { + // clear the flag + Thread.interrupted(); + } + } + + @Test + void isClosed_eofThenReset_returnsFalse() throws IOException { + when(mockStream.read()).thenReturn(-1); + filterInputStream.read(); + assertThat(filterInputStream.isClosed()).isTrue(); + + filterInputStream.reset(); + + assertThat(filterInputStream.isClosed()).isFalse(); + } + + @Test + void isClosed_closedThenReset_resetThrows_returnsTrue() throws IOException { + when(mockStream.read()).thenReturn(-1); + filterInputStream.read(); + assertThat(filterInputStream.isClosed()).isTrue(); + + doThrow(new RuntimeException("no reset")).when(mockStream).reset(); + assertThatThrownBy(filterInputStream::reset).isInstanceOf(RuntimeException.class); + + assertThat(filterInputStream.isClosed()).isTrue(); + } + + private static class TestInputStream extends SdkFilterInputStream { + + protected TestInputStream(InputStream in) { + super(in); + } + + @Override + public boolean isClosed() { + return super.isClosed(); + } + } +}