[flink] branch release-1.17 updated: [FLINK-31708][API/Type Serialization System] Make DataInputView#read(byte[], int, int) return 0 if len is 0.

[gaoyunhaii]

This is an automated email from the ASF dual-hosted git repository.

gaoyunhaii pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.17 by this push:
     new 729043df687 [FLINK-31708][API/Type Serialization System] Make 
DataInputView#read(byte[], int, int) return 0 if len is 0.
729043df687 is described below

commit 729043df687a96711d3591fcdf5e8e712cd21b87
Author: 沈嘉琦 <shenji...@sensorsdata.cn>
AuthorDate: Mon Apr 3 19:43:14 2023 +0800

    [FLINK-31708][API/Type Serialization System] Make 
DataInputView#read(byte[], int, int) return 0 if len is 0.
    
    This closes #22335.
---
 .../org/apache/flink/core/memory/DataInputDeserializer.java    |  2 +-
 .../main/java/org/apache/flink/core/memory/DataInputView.java  |  7 ++++++-
 flink-core/src/main/java/org/apache/flink/types/Record.java    |  2 +-
 .../apache/flink/core/memory/DataInputDeserializerTest.java    | 10 ++++++++++
 4 files changed, 18 insertions(+), 3 deletions(-)

diff --git 
a/flink-core/src/main/java/org/apache/flink/core/memory/DataInputDeserializer.java
 
b/flink-core/src/main/java/org/apache/flink/core/memory/DataInputDeserializer.java
index 9ece9f00135..051bad2650e 100644
--- 
a/flink-core/src/main/java/org/apache/flink/core/memory/DataInputDeserializer.java
+++ 
b/flink-core/src/main/java/org/apache/flink/core/memory/DataInputDeserializer.java
@@ -378,7 +378,7 @@ public class DataInputDeserializer implements 
DataInputView, java.io.Serializabl
         }
 
         if (this.position >= this.end) {
-            return -1;
+            return len == 0 ? 0 : -1;
         } else {
             int toRead = Math.min(this.end - this.position, len);
             System.arraycopy(this.buffer, this.position, b, off, toRead);
diff --git 
a/flink-core/src/main/java/org/apache/flink/core/memory/DataInputView.java 
b/flink-core/src/main/java/org/apache/flink/core/memory/DataInputView.java
index 43f34f5383f..58d5c634aed 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/DataInputView.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/DataInputView.java
@@ -44,7 +44,12 @@ public interface DataInputView extends DataInput {
 
     /**
      * Reads up to {@code len} bytes of memory and stores it into {@code b} 
starting at offset
-     * {@code off}. It returns the number of read bytes or -1 if there is no 
more data left.
+     * {@code off}.
+     *
+     * <p>If <code>len</code> is zero, then no bytes are read and 
<code>0</code> is returned;
+     * otherwise, there is an attempt to read at least one byte. If there is 
no more data left, the
+     * value <code>-1</code> is returned; otherwise, at least one byte is read 
and stored into
+     * <code>b</code>.
      *
      * @param b byte array to store the data to
      * @param off offset into byte array
diff --git a/flink-core/src/main/java/org/apache/flink/types/Record.java 
b/flink-core/src/main/java/org/apache/flink/types/Record.java
index 59e1568fc8f..a7b6f989b7d 100644
--- a/flink-core/src/main/java/org/apache/flink/types/Record.java
+++ b/flink-core/src/main/java/org/apache/flink/types/Record.java
@@ -1584,7 +1584,7 @@ public final class Record implements Value, 
CopyableValue<Record> {
             }
 
             if (this.position >= this.end) {
-                return -1;
+                return len == 0 ? 0 : -1;
             } else {
                 int toRead = Math.min(this.end - this.position, len);
                 System.arraycopy(this.memory, this.position, b, off, toRead);
diff --git 
a/flink-core/src/test/java/org/apache/flink/core/memory/DataInputDeserializerTest.java
 
b/flink-core/src/test/java/org/apache/flink/core/memory/DataInputDeserializerTest.java
index d8ebd5968cb..e8707f1eb70 100644
--- 
a/flink-core/src/test/java/org/apache/flink/core/memory/DataInputDeserializerTest.java
+++ 
b/flink-core/src/test/java/org/apache/flink/core/memory/DataInputDeserializerTest.java
@@ -54,4 +54,14 @@ public class DataInputDeserializerTest {
         }
         Assert.assertEquals(0, dis.available());
     }
+
+    @Test
+    public void testReadWithLenZero() throws IOException {
+        byte[] bytes = new byte[0];
+        DataInputDeserializer dis = new DataInputDeserializer(bytes, 0, 
bytes.length);
+        Assert.assertEquals(0, dis.available());
+
+        byte[] bytesForRead = new byte[0];
+        Assert.assertEquals(0, dis.read(bytesForRead, 0, 0)); // do not throw 
when read with len 0
+    }
 }