fix ByteStreamStateHandle#read(byte[] bytes, int offset, int count) r…

…eturn -1 when count = 0 and no data left.
BugsEverywhere · Oct 26, 2022 · db46d43 · db46d43
1 parent 93af1e4
commit db46d43
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 1 deletion.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/ByteStreamStateHandle.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/ByteStreamStateHandle.java
@@ -148,7 +148,7 @@ public int read(byte[] b, int off, int len) throws IOException {
                 index += bytesToCopy;
                 return bytesToCopy;
             } else {
-                return -1;
+                return len == 0 ? 0 : -1;
             }
         }
     }

diff --git a/...untime/src/test/java/org/apache/flink/runtime/state/memory/ByteStreamStateHandleTest.java b/...untime/src/test/java/org/apache/flink/runtime/state/memory/ByteStreamStateHandleTest.java
@@ -172,4 +172,15 @@ public void testBulkReadINdexOutOfBounds() throws IOException {
             // expected
         }
     }
+
+    @Test
+    public void testStreamWithEmptyByteArray() throws IOException {
+        final byte[] data = new byte[0];
+        final ByteStreamStateHandle handle = new ByteStreamStateHandle("name", data);
+
+        FSDataInputStream in = handle.openInputStream();
+        in.seek(0);
+        byte[] dataGot = new byte[1];
+        assertEquals(0, in.read(dataGot, 0, 0));
+    }
 }