This is an automated email from the ASF dual-hosted git repository.
yuzelin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new f08861706 [flink] Unified use of JsonSerdeUtil (#2747)
f08861706 is described below
commit f0886170601a7af2de5a933c56bfa68be4341bd7
Author: monster <[email protected]>
AuthorDate: Tue Jan 23 10:31:20 2024 +0800
[flink] Unified use of JsonSerdeUtil (#2747)
---
.../src/main/java/org/apache/paimon/utils/JsonSerdeUtil.java | 8 ++++++++
.../apache/paimon/flink/action/cdc/format/RecordParser.java | 10 +++-------
.../flink/action/cdc/format/canal/CanalRecordParser.java | 5 +++--
3 files changed, 14 insertions(+), 9 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/utils/JsonSerdeUtil.java
b/paimon-core/src/main/java/org/apache/paimon/utils/JsonSerdeUtil.java
index 19a57ac58..d45a93368 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/JsonSerdeUtil.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/JsonSerdeUtil.java
@@ -233,6 +233,14 @@ public class JsonSerdeUtil {
return true;
}
+ public static <T> T convertValue(Object fromValue, TypeReference<T>
toValueTypeRef) {
+ return OBJECT_MAPPER_INSTANCE.convertValue(fromValue, toValueTypeRef);
+ }
+
+ public static String writeValueAsString(Object value) throws
JsonProcessingException {
+ return OBJECT_MAPPER_INSTANCE.writer().writeValueAsString(value);
+ }
+
public static boolean isNull(JsonNode jsonNode) {
return jsonNode == null || jsonNode.isNull();
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParser.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParser.java
index 6903a6707..66d313135 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParser.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParser.java
@@ -32,7 +32,6 @@ import org.apache.paimon.utils.TypeUtils;
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.type.TypeReference;
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
-import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.ArrayNode;
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
@@ -73,8 +72,6 @@ public abstract class RecordParser implements
FlatMapFunction<String, RichCdcMul
protected static final String FIELD_TABLE = "table";
protected static final String FIELD_DATABASE = "database";
- protected static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
-
private final boolean caseSensitive;
protected final TypeMapping typeMapping;
protected final List<ComputedColumn> computedColumns;
@@ -143,7 +140,7 @@ public abstract class RecordParser implements
FlatMapFunction<String, RichCdcMul
JsonNode record, LinkedHashMap<String, DataType> paimonFieldTypes)
{
paimonFieldTypes.putAll(fillDefaultStringTypes(record));
Map<String, Object> recordMap =
- OBJECT_MAPPER.convertValue(record, new
TypeReference<Map<String, Object>>() {});
+ JsonSerdeUtil.convertValue(record, new
TypeReference<Map<String, Object>>() {});
Map<String, String> rowData =
recordMap.entrySet().stream()
.collect(
@@ -153,9 +150,8 @@ public abstract class RecordParser implements
FlatMapFunction<String, RichCdcMul
if
(Objects.nonNull(entry.getValue())
&&
!TypeUtils.isBasicType(entry.getValue())) {
try {
- return OBJECT_MAPPER
- .writer()
-
.writeValueAsString(entry.getValue());
+ return
JsonSerdeUtil.writeValueAsString(
+ entry.getValue());
} catch
(JsonProcessingException e) {
LOG.error("Failed to
deserialize record.", e);
return
Objects.toString(entry.getValue());
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/canal/CanalRecordParser.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/canal/CanalRecordParser.java
index 3ea979a8f..2877b0ed3 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/canal/CanalRecordParser.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/canal/CanalRecordParser.java
@@ -25,6 +25,7 @@ import
org.apache.paimon.flink.action.cdc.mysql.MySqlTypeUtils;
import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.RowKind;
+import org.apache.paimon.utils.JsonSerdeUtil;
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.type.TypeReference;
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
@@ -135,7 +136,7 @@ public class CanalRecordParser extends RecordParser {
return null;
}
- return OBJECT_MAPPER.convertValue(
+ return JsonSerdeUtil.convertValue(
schema, new TypeReference<LinkedHashMap<String, String>>() {});
}
@@ -163,7 +164,7 @@ public class CanalRecordParser extends RecordParser {
JsonNode record, LinkedHashMap<String, DataType> paimonFieldTypes)
{
LinkedHashMap<String, String> originalFieldTypes =
tryExtractOriginalFieldTypes();
Map<String, Object> recordMap =
- OBJECT_MAPPER.convertValue(record, new
TypeReference<Map<String, Object>>() {});
+ JsonSerdeUtil.convertValue(record, new
TypeReference<Map<String, Object>>() {});
Map<String, String> rowData = new HashMap<>();
if (originalFieldTypes != null) {
