From 070ce04dbba08461b72dca4753fc1d757d511615 Mon Sep 17 00:00:00 2001 From: emhui Date: Thu, 27 Jul 2023 21:19:10 +0800 Subject: [PATCH 1/2] [INLONG-8601][Sort] Fix NPE in schema.name of Oracle CDC --- .../table/RowDataDebeziumDeserializeSchema.java | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/table/RowDataDebeziumDeserializeSchema.java b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/table/RowDataDebeziumDeserializeSchema.java index 6e001ec6812..a6b0ed261c8 100644 --- a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/table/RowDataDebeziumDeserializeSchema.java +++ b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/table/RowDataDebeziumDeserializeSchema.java @@ -17,6 +17,7 @@ package org.apache.inlong.sort.cdc.oracle.debezium.table; +import org.apache.commons.lang.StringUtils; import org.apache.inlong.sort.cdc.base.debezium.DebeziumDeserializationSchema; import org.apache.inlong.sort.cdc.base.debezium.table.AppendMetadataCollector; import org.apache.inlong.sort.cdc.base.debezium.table.DeserializationRuntimeConverter; @@ -292,7 +293,7 @@ private static DeserializationRuntimeConverter convertToTime() { @Override public Object convert(Object dbzObj, Schema schema) { - if (dbzObj instanceof Long) { + if (dbzObj instanceof Long && StringUtils.isNotBlank(schema.name())) { // Because Oracle CDC has been shaded, the schema will have the prefix // 'org.apache.inlong.sort.cdc.oracle.shaded' added, // so we need to use `schemaName.endsWith()` to determine the Schema type. @@ -327,7 +328,7 @@ private static DeserializationRuntimeConverter convertToTimestamp(ZoneId serverT @Override public Object convert(Object dbzObj, Schema schema) { - if (dbzObj instanceof Long) { + if (dbzObj instanceof Long && StringUtils.isNotBlank(schema.name())) { // Because Oracle CDC has been shaded, the schema will have the prefix // 'org.apache.inlong.sort.cdc.oracle.shaded' added, // so we need to use `schemaName.endsWith()` to determine the Schema type. @@ -462,7 +463,8 @@ public Object convert(Object dbzObj, Schema schema) { // Because Oracle CDC has been shaded, the schema will have the prefix // 'org.apache.inlong.sort.cdc.oracle.shaded' added, // so we need to use `schemaName.endsWith()` to determine the Schema type. - if (schema.name().endsWith(VariableScaleDecimal.LOGICAL_NAME)) { + if (StringUtils.isNotBlank(schema.name()) + && schema.name().endsWith(VariableScaleDecimal.LOGICAL_NAME)) { SpecialValueDecimal decimal = VariableScaleDecimal.toLogical((Struct) dbzObj); bigDecimal = decimal.getDecimalValue().orElse(BigDecimal.ZERO); @@ -768,8 +770,8 @@ public Object convert(Object dbzObj, Schema schema, TableChange tableSchema) thr * @return the extracted data with schema */ private Object getValueWithSchema(Object fieldValue, String schemaName) { - if (fieldValue == null) { - return null; + if (fieldValue == null || StringUtils.isBlank(schemaName)) { + return fieldValue; } // Because Oracle CDC has been shaded, the schema will have the prefix // 'org.apache.inlong.sort.cdc.oracle.shaded' added, From e8e4b1f96ed04fb26e07d0338b5adbde24fe4684 Mon Sep 17 00:00:00 2001 From: e-mhui Date: Fri, 28 Jul 2023 12:30:14 +0800 Subject: [PATCH 2/2] [INLONG-8594][Sort] Fix checkstyle --- .../oracle/debezium/table/RowDataDebeziumDeserializeSchema.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/table/RowDataDebeziumDeserializeSchema.java b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/table/RowDataDebeziumDeserializeSchema.java index a6b0ed261c8..d7dd5081a25 100644 --- a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/table/RowDataDebeziumDeserializeSchema.java +++ b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/table/RowDataDebeziumDeserializeSchema.java @@ -17,7 +17,6 @@ package org.apache.inlong.sort.cdc.oracle.debezium.table; -import org.apache.commons.lang.StringUtils; import org.apache.inlong.sort.cdc.base.debezium.DebeziumDeserializationSchema; import org.apache.inlong.sort.cdc.base.debezium.table.AppendMetadataCollector; import org.apache.inlong.sort.cdc.base.debezium.table.DeserializationRuntimeConverter; @@ -38,6 +37,7 @@ import io.debezium.time.NanoTimestamp; import io.debezium.time.Timestamp; import io.debezium.time.ZonedTimestamp; +import org.apache.commons.lang.StringUtils; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.table.data.DecimalData; import org.apache.flink.table.data.GenericRowData;