[flink] branch release-1.12 updated: [FLINK-23120][python] Fix ByteArrayWrapperSerializer.serialize to use writeInt to serialize the length

[dianfu]

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

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


The following commit(s) were added to refs/heads/release-1.12 by this push:
     new d77c51d  [FLINK-23120][python] Fix 
ByteArrayWrapperSerializer.serialize to use writeInt to serialize the length
d77c51d is described below

commit d77c51d449e8d25611b0d84fe5d438949f360384
Author: Dian Fu <dia...@apache.org>
AuthorDate: Wed Jun 23 17:20:20 2021 +0800

    [FLINK-23120][python] Fix ByteArrayWrapperSerializer.serialize to use 
writeInt to serialize the length
    
    This closes #16258.
---
 .../api/utils/ByteArrayWrapperSerializer.java      | 12 ++--
 .../api/utils/ByteArrayWrapperSerializerTest.java  | 64 ++++++++++++++++++++++
 2 files changed, 71 insertions(+), 5 deletions(-)

diff --git 
a/flink-python/src/main/java/org/apache/flink/streaming/api/utils/ByteArrayWrapperSerializer.java
 
b/flink-python/src/main/java/org/apache/flink/streaming/api/utils/ByteArrayWrapperSerializer.java
index 96aaab2..f3bea10 100644
--- 
a/flink-python/src/main/java/org/apache/flink/streaming/api/utils/ByteArrayWrapperSerializer.java
+++ 
b/flink-python/src/main/java/org/apache/flink/streaming/api/utils/ByteArrayWrapperSerializer.java
@@ -68,7 +68,7 @@ public class ByteArrayWrapperSerializer extends 
TypeSerializerSingleton<ByteArra
 
     @Override
     public void serialize(ByteArrayWrapper record, DataOutputView target) 
throws IOException {
-        target.write(record.getLimit() - record.getOffset());
+        target.writeInt(record.getLimit() - record.getOffset());
         target.write(record.getData(), record.getOffset(), record.getLimit() - 
record.getOffset());
     }
 
@@ -76,7 +76,7 @@ public class ByteArrayWrapperSerializer extends 
TypeSerializerSingleton<ByteArra
     public ByteArrayWrapper deserialize(DataInputView source) throws 
IOException {
         int length = source.readInt();
         byte[] result = new byte[length];
-        source.read(result);
+        source.readFully(result);
         return new ByteArrayWrapper(result);
     }
 
@@ -85,8 +85,10 @@ public class ByteArrayWrapperSerializer extends 
TypeSerializerSingleton<ByteArra
             throws IOException {
         int length = source.readInt();
         byte[] result = new byte[length];
-        source.read(result);
+        source.readFully(result);
         reuse.setData(result);
+        reuse.setOffset(0);
+        reuse.setLimit(result.length);
         return reuse;
     }
 
@@ -94,8 +96,8 @@ public class ByteArrayWrapperSerializer extends 
TypeSerializerSingleton<ByteArra
     public void copy(DataInputView source, DataOutputView target) throws 
IOException {
         int length = source.readInt();
         byte[] result = new byte[length];
-        source.read(result);
-        target.write(length);
+        source.readFully(result);
+        target.writeInt(length);
         target.write(result);
     }
 
diff --git 
a/flink-python/src/test/java/org/apache/flink/streaming/api/utils/ByteArrayWrapperSerializerTest.java
 
b/flink-python/src/test/java/org/apache/flink/streaming/api/utils/ByteArrayWrapperSerializerTest.java
new file mode 100644
index 0000000..5547ad8
--- /dev/null
+++ 
b/flink-python/src/test/java/org/apache/flink/streaming/api/utils/ByteArrayWrapperSerializerTest.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 org.apache.flink.streaming.api.utils;
+
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+
+import java.util.Random;
+
+/** Test class for {@link ByteArrayWrapperSerializer}. */
+public class ByteArrayWrapperSerializerTest extends 
SerializerTestBase<ByteArrayWrapper> {
+
+    private final Random rnd = new Random(346283764872L);
+
+    private static final ByteArrayWrapper EMPTY_ARRAY = new 
ByteArrayWrapper(new byte[] {});
+
+    @Override
+    protected TypeSerializer<ByteArrayWrapper> createSerializer() {
+        return new ByteArrayWrapperSerializer();
+    }
+
+    @Override
+    protected Class<ByteArrayWrapper> getTypeClass() {
+        return ByteArrayWrapper.class;
+    }
+
+    @Override
+    protected int getLength() {
+        return -1;
+    }
+
+    @Override
+    protected ByteArrayWrapper[] getTestData() {
+        return new ByteArrayWrapper[] {
+            EMPTY_ARRAY,
+            randomByteArray(rnd.nextInt(1024 * 1024)),
+            randomByteArray(1),
+            randomByteArray(2),
+            randomByteArray(1024 * 1024),
+            randomByteArray(32 * 1024 * 1024),
+        };
+    }
+
+    private ByteArrayWrapper randomByteArray(int len) {
+        byte[] data = new byte[len];
+        rnd.nextBytes(data);
+        return new ByteArrayWrapper(data);
+    }
+}