[flink] branch release-1.11 updated: [FLINK-19790][json] Clear reused ObjectNode's content for map converter in RowDataToJsonConverters

[libenchao]

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

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


The following commit(s) were added to refs/heads/release-1.11 by this push:
     new b50fc62  [FLINK-19790][json] Clear reused ObjectNode's content for map 
converter in RowDataToJsonConverters
b50fc62 is described below

commit b50fc62e5b08363932815e0c68fbea3f08f0011f
Author: Benchao Li <libenc...@gmail.com>
AuthorDate: Wed Nov 4 21:33:23 2020 +0800

    [FLINK-19790][json] Clear reused ObjectNode's content for map converter in 
RowDataToJsonConverters
    
    This closes 13926
---
 .../json/JsonRowDataSerializationSchema.java       |  1 +
 .../formats/json/JsonRowDataSerDeSchemaTest.java   | 23 +++++++++++++++++++++-
 2 files changed, 23 insertions(+), 1 deletion(-)

diff --git 
a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDataSerializationSchema.java
 
b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDataSerializationSchema.java
index 89b3b87..afc8ede 100644
--- 
a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDataSerializationSchema.java
+++ 
b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDataSerializationSchema.java
@@ -268,6 +268,7 @@ public class JsonRowDataSerializationSchema implements 
SerializationSchema<RowDa
                                node = mapper.createObjectNode();
                        } else {
                                node = (ObjectNode) reuse;
+                               node.removeAll();
                        }
 
                        MapData map = (MapData) object;
diff --git 
a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDataSerDeSchemaTest.java
 
b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDataSerDeSchemaTest.java
index fedfbea..a27de49 100644
--- 
a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDataSerDeSchemaTest.java
+++ 
b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDataSerDeSchemaTest.java
@@ -232,7 +232,12 @@ public class JsonRowDataSerDeSchemaTest {
                RowType rowType = (RowType) ROW(
                        FIELD("f1", INT()),
                        FIELD("f2", BOOLEAN()),
-                       FIELD("f3", STRING())
+                       FIELD("f3", STRING()),
+                       FIELD("f4", MAP(STRING(), STRING())),
+                       FIELD("f5", ARRAY(STRING())),
+                       FIELD("f6", ROW(
+                               FIELD("f1", STRING()),
+                               FIELD("f2", INT())))
                ).getLogicalType();
 
                JsonRowDataDeserializationSchema deserializationSchema = new 
JsonRowDataDeserializationSchema(
@@ -247,6 +252,14 @@ public class JsonRowDataSerDeSchemaTest {
                        root.put("f1", 1);
                        root.put("f2", true);
                        root.put("f3", "str");
+                       ObjectNode map = root.putObject("f4");
+                       map.put("hello1", "flink");
+                       ArrayNode array = root.putArray("f5");
+                       array.add("element1");
+                       array.add("element2");
+                       ObjectNode row = root.putObject("f6");
+                       row.put("f1", "this is row1");
+                       row.put("f2", 12);
                        byte[] serializedJson = 
objectMapper.writeValueAsBytes(root);
                        RowData rowData = 
deserializationSchema.deserialize(serializedJson);
                        byte[] actual = serializationSchema.serialize(rowData);
@@ -259,6 +272,14 @@ public class JsonRowDataSerDeSchemaTest {
                        root.put("f1", 10);
                        root.put("f2", false);
                        root.put("f3", "newStr");
+                       ObjectNode map = root.putObject("f4");
+                       map.put("hello2", "json");
+                       ArrayNode array = root.putArray("f5");
+                       array.add("element3");
+                       array.add("element4");
+                       ObjectNode row = root.putObject("f6");
+                       row.put("f1", "this is row2");
+                       row.putNull("f2");
                        byte[] serializedJson = 
objectMapper.writeValueAsBytes(root);
                        RowData rowData = 
deserializationSchema.deserialize(serializedJson);
                        byte[] actual = serializationSchema.serialize(rowData);