InputStream#readAllBytes() should receive no premature EOS (#4258)

* InputStream#readAllBytes() should receive no premature EOS

A read-call with len == 0 happens here:
https://github.com/AdoptOpenJDK/openjdk-jdk11/blob/master/src/java.base/share/classes/java/io/InputStream.java#L396

* Add/Fix tests

Fixed some of the tests since the assertions fail if the read length is 0. This
is in line with the default implementation of `read(byte[],int,int)` in
`InputStream`:
https://github.com/openjdk/jdk11u/blob/d77215acdd6b9008d7f58c7ad5a82d6087c20f86/src/java.base/share/classes/java/io/InputStream.java#L267

* Add changelog entry

---------

Co-authored-by: Dongie Agnir <dongie@amazon.com>
Co-authored-by: Dongie Agnir <261310+dagnir@users.noreply.github.com>

.changes/next-release/bugfix-AWSSDKforJavav2-2984eeb.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS SDK for Java v2",
+    "contributor": "faucct",
+    "description": "Fix `InputStreamSubscriber` to return 0 when reading a length of 0."
+}

utils/src/main/java/software/amazon/awssdk/utils/async/InputStreamSubscriber.java

@@ -102,6 +102,10 @@ public int read(byte[] b) {
 
     @Override
     public int read(byte[] bytes, int off, int len) {
+        if (len == 0) {
+            return 0;
+        }
+
         ByteBuffer byteBuffer = ByteBuffer.wrap(bytes, off, len);
         TransferResult transferResult = delegate.blockingTransferTo(byteBuffer);
         int dataTransferred = byteBuffer.position() - off;

utils/src/test/java/software/amazon/awssdk/utils/async/InputStreamSubscriberTest.java

@@ -58,8 +58,8 @@ public void onComplete_returnsEndOfStream_onRead() {
         publisher.subscribe(subscriber);
         publisher.complete();
         assertThat(subscriber.read()).isEqualTo(-1);
-        assertThat(subscriber.read(new byte[0])).isEqualTo(-1);
-        assertThat(subscriber.read(new byte[0], 0, 0)).isEqualTo(-1);
+        assertThat(subscriber.read(new byte[1])).isEqualTo(-1);
+        assertThat(subscriber.read(new byte[1], 0, 1)).isEqualTo(-1);
     }
 
     @Test
@@ -69,8 +69,8 @@ public void onError_throws_onRead() {
         publisher.subscribe(subscriber);
         publisher.error(exception);
         assertThatThrownBy(() -> subscriber.read()).isEqualTo(exception);
-        assertThatThrownBy(() -> subscriber.read(new byte[0])).isEqualTo(exception);
-        assertThatThrownBy(() -> subscriber.read(new byte[0], 0, 0)).isEqualTo(exception);
+        assertThatThrownBy(() -> subscriber.read(new byte[1])).isEqualTo(exception);
+        assertThatThrownBy(() -> subscriber.read(new byte[1], 0, 1)).isEqualTo(exception);
     }
 
     @Test
@@ -128,8 +128,15 @@ public void read_afterClose_fails() {
         publisher.subscribe(subscriber);
         subscriber.close();
         assertThatThrownBy(() -> subscriber.read()).isInstanceOf(CancellationException.class);
-        assertThatThrownBy(() -> subscriber.read(new byte[0])).isInstanceOf(CancellationException.class);
-        assertThatThrownBy(() -> subscriber.read(new byte[0], 0, 0)).isInstanceOf(CancellationException.class);
+        assertThatThrownBy(() -> subscriber.read(new byte[1])).isInstanceOf(CancellationException.class);
+        assertThatThrownBy(() -> subscriber.read(new byte[1], 0, 1)).isInstanceOf(CancellationException.class);
+    }
+
+    @Test
+    public void readByteArray_0Len_returns0() {
+        publisher.subscribe(subscriber);
+
+        assertThat(subscriber.read(new byte[1], 0, 0)).isEqualTo(0);
     }
 
     public static List<Arguments> stochastic_methodCallsSeemThreadSafe_parameters() {