This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 0b1d75fc66 [common][cdc] Fail loudly on malformed complex-type JSON
instead of silently returning empty (#8708)
0b1d75fc66 is described below
commit 0b1d75fc66bf549899daaf61bf70161ef5a796bc
Author: Vova Kolmakov <[email protected]>
AuthorDate: Sat Jul 18 11:05:19 2026 +0700
[common][cdc] Fail loudly on malformed complex-type JSON instead of
silently returning empty (#8708)
---
.../java/org/apache/paimon/utils/TypeUtils.java | 30 +++---
.../org/apache/paimon/utils/TypeUtilsTest.java | 89 ++++++++++++++++
.../paimon/flink/sink/cdc/CdcRecordUtils.java | 2 -
.../sink/cdc/CdcRecordStoreWriteOperatorTest.java | 116 +++++++++++++++++++++
4 files changed, 223 insertions(+), 14 deletions(-)
diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/TypeUtils.java
b/paimon-common/src/main/java/org/apache/paimon/utils/TypeUtils.java
index bcb3f91b76..cfe0f95afb 100644
--- a/paimon-common/src/main/java/org/apache/paimon/utils/TypeUtils.java
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/TypeUtils.java
@@ -51,7 +51,6 @@ import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Base64;
-import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -173,6 +172,12 @@ public class TypeUtils {
DataType elementType = arrayType.getElementType();
try {
JsonNode arrayNode = OBJECT_MAPPER.readTree(s);
+ if (!arrayNode.isArray()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Expected a JSON array for type %s,
but got %s",
+ type, arrayNode.getNodeType()));
+ }
List<Object> resultList = new ArrayList<>();
for (JsonNode elementNode : arrayNode) {
if (!elementNode.isNull()) {
@@ -236,6 +241,12 @@ public class TypeUtils {
DataType valueType = mapType.getValueType();
try {
JsonNode mapNode = OBJECT_MAPPER.readTree(s);
+ if (!mapNode.isObject()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Expected a JSON object for type %s,
but got %s",
+ type, mapNode.getNodeType()));
+ }
Map<Object, Object> resultMap = new HashMap<>();
mapNode.fields()
.forEachRemaining(
@@ -262,11 +273,6 @@ public class TypeUtils {
resultMap.put(key, value);
});
return new GenericMap(resultMap);
- } catch (JsonProcessingException e) {
- LOG.info(
- String.format("Failed to parse MAP for type %s
with value %s", type, s),
- e);
- return new GenericMap(Collections.emptyMap());
} catch (Exception e) {
throw new RuntimeException(
String.format("Failed to parse Json String %s",
s), e);
@@ -275,6 +281,12 @@ public class TypeUtils {
RowType rowType = (RowType) type;
try {
JsonNode rowNode = OBJECT_MAPPER.readTree(s);
+ if (!rowNode.isObject()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Expected a JSON object for type %s,
but got %s",
+ type, rowNode.getNodeType()));
+ }
GenericRow genericRow =
new GenericRow(
rowType.getFields()
@@ -297,12 +309,6 @@ public class TypeUtils {
}
}
return genericRow;
- } catch (JsonProcessingException e) {
- LOG.info(
- String.format(
- "Failed to parse ROW for type %s with
value %s", type, s),
- e);
- return new GenericRow(0);
} catch (Exception e) {
throw new RuntimeException(
String.format("Failed to parse Json String %s",
s), e);
diff --git
a/paimon-common/src/test/java/org/apache/paimon/utils/TypeUtilsTest.java
b/paimon-common/src/test/java/org/apache/paimon/utils/TypeUtilsTest.java
index 54c1af27d8..083d08f485 100644
--- a/paimon-common/src/test/java/org/apache/paimon/utils/TypeUtilsTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/utils/TypeUtilsTest.java
@@ -26,11 +26,14 @@ import org.apache.paimon.data.GenericMap;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.Timestamp;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
@@ -42,6 +45,7 @@ import java.util.HashMap;
import java.util.TimeZone;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link TypeUtils}. */
public class TypeUtilsTest {
@@ -298,6 +302,91 @@ public class TypeUtilsTest {
assertThat(result).isEqualTo(expected);
}
+ private static DataType twoStringFieldRow() {
+ return DataTypes.ROW(
+ new DataField(0, "key1", DataTypes.STRING()),
+ new DataField(1, "key2", DataTypes.STRING()));
+ }
+
+ @Test
+ public void testMapCastFromMalformedJsonThrows() {
+ assertThatThrownBy(
+ () ->
+ TypeUtils.castFromString(
+ "{\"a\": ",
+ DataTypes.MAP(DataTypes.STRING(),
DataTypes.STRING())))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to parse Json String");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"[1,2]", "123", "\"abc\"", "null", ""})
+ public void testMapCastFromNonObjectJsonThrows(String value) {
+ assertThatThrownBy(
+ () ->
+ TypeUtils.castFromString(
+ value,
+ DataTypes.MAP(DataTypes.STRING(),
DataTypes.STRING())))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to parse Json String");
+ }
+
+ @Test
+ public void testMapCastFromEmptyObject() {
+ Object result =
+ TypeUtils.castFromString(
+ "{}", DataTypes.MAP(DataTypes.STRING(),
DataTypes.STRING()));
+ assertThat(result).isEqualTo(new GenericMap(Collections.emptyMap()));
+ }
+
+ @Test
+ public void testRowCastFromMalformedJsonThrows() {
+ assertThatThrownBy(() -> TypeUtils.castFromString("{\"key1\": ",
twoStringFieldRow()))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to parse Json String");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"[1,2]", "123", "\"abc\"", "null", ""})
+ public void testRowCastFromNonObjectJsonThrows(String value) {
+ assertThatThrownBy(() -> TypeUtils.castFromString(value,
twoStringFieldRow()))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to parse Json String");
+ }
+
+ /** A CDC record may legitimately not carry every field, so missing fields
stay tolerated. */
+ @Test
+ public void testRowCastFromObjectWithMissingField() {
+ Object result = TypeUtils.castFromString("{\"key2\":\"v\"}",
twoStringFieldRow());
+ assertThat(result).isEqualTo(GenericRow.of(null,
BinaryString.fromString("v")));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"123", "{\"a\":1}", ""})
+ public void testArrayCastFromNonArrayJsonThrows(String value) {
+ assertThatThrownBy(() -> TypeUtils.castFromString(value,
DataTypes.ARRAY(DataTypes.INT())))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to parse Json String");
+ assertThatThrownBy(
+ () -> TypeUtils.castFromString(value,
DataTypes.ARRAY(DataTypes.STRING())))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to parse Json String");
+ }
+
+ /** Values that are not JSON at all still fall back to the legacy
comma-separated parsing. */
+ @Test
+ public void testArrayCastFromLegacyCommaSeparated() {
+ Object result = TypeUtils.castFromString("a,b,c",
DataTypes.ARRAY(DataTypes.STRING()));
+ GenericArray expected =
+ new GenericArray(
+ Arrays.asList(
+ BinaryString.fromString("a"),
+ BinaryString.fromString("b"),
+ BinaryString.fromString("c"))
+ .toArray());
+ assertThat(result).isEqualTo(expected);
+ }
+
@Test
public void testSmallIntCastFromString() {
String value = "12";
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordUtils.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordUtils.java
index 503302b9b0..6fc94eb9e5 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordUtils.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordUtils.java
@@ -100,8 +100,6 @@ public class CdcRecordUtils {
}
DataType type = dataFields.get(idx).type();
- // TODO TypeUtils.castFromString cannot deal with complex types
like arrays and
- // maps. Change type of CdcRecord#field if needed.
try {
genericRow.setField(idx,
TypeUtils.castFromCdcValueString(value, type));
} catch (Exception e) {
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreWriteOperatorTest.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreWriteOperatorTest.java
index d27b919fa1..7fa0b09de3 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreWriteOperatorTest.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreWriteOperatorTest.java
@@ -32,6 +32,8 @@ import org.apache.paimon.schema.SchemaUtils;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowKind;
@@ -62,6 +64,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Predicate;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link CdcRecordStoreWriteOperator}. */
public class CdcRecordStoreWriteOperatorTest {
@@ -251,6 +254,109 @@ public class CdcRecordStoreWriteOperatorTest {
harness.close();
}
+ private static RowType rowTypeWithComplexColumn(DataType complexType) {
+ return RowType.of(new DataType[] {DataTypes.INT(), complexType}, new
String[] {"k", "v"});
+ }
+
+ private CdcRecord recordWithValue(String value) {
+ Map<String, String> data = new HashMap<>();
+ data.put("k", "1");
+ data.put("v", value);
+ return new CdcRecord(RowKind.INSERT, data);
+ }
+
+ /**
+ * A malformed MAP value must reach the corrupt record policy, the same
way a malformed value of
+ * any other type does, instead of being silently written as an empty map.
+ */
+ @Test
+ @Timeout(30)
+ public void testCorruptMapRecordFailsLoudly() throws Exception {
+ Options options = new Options();
+ options.set(CdcRecordStoreWriteOperator.MAX_RETRY_NUM_TIMES, 1);
+ FileStoreTable table =
+ createFileStoreTable(
+ rowTypeWithComplexColumn(
+ DataTypes.MAP(DataTypes.STRING(),
DataTypes.STRING())),
+ Collections.emptyList(),
+ Collections.singletonList("k"),
+ options);
+
+ OneInputStreamOperatorTestHarness<CdcRecord, Committable> harness =
+ createTestHarness(table);
+ harness.open();
+
+ assertThatThrownBy(() ->
harness.processElement(recordWithValue("{\"a\": "), 1))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Unable to process element. Possibly a
corrupt record");
+
+ harness.close();
+ }
+
+ /** Same for ROW, which used to blow up later with an obscure error while
writing the row. */
+ @Test
+ @Timeout(30)
+ public void testCorruptRowRecordFailsLoudly() throws Exception {
+ Options options = new Options();
+ options.set(CdcRecordStoreWriteOperator.MAX_RETRY_NUM_TIMES, 1);
+ FileStoreTable table =
+ createFileStoreTable(
+ rowTypeWithComplexColumn(
+ DataTypes.ROW(
+ new DataField(2, "f0",
DataTypes.STRING()),
+ new DataField(3, "f1",
DataTypes.STRING()))),
+ Collections.emptyList(),
+ Collections.singletonList("k"),
+ options);
+
+ OneInputStreamOperatorTestHarness<CdcRecord, Committable> harness =
+ createTestHarness(table);
+ harness.open();
+
+ assertThatThrownBy(() ->
harness.processElement(recordWithValue("{\"f0\": "), 1))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Unable to process element. Possibly a
corrupt record");
+
+ harness.close();
+ }
+
+ /**
+ * With {@code cdc.skip-corrupt-record} the record must be dropped
entirely, rather than written
+ * with an empty map in place of the value that failed to parse.
+ */
+ @Test
+ @Timeout(30)
+ public void testSkipCorruptMapRecord() throws Exception {
+ Options options = new Options();
+ options.set(CdcRecordStoreWriteOperator.MAX_RETRY_NUM_TIMES, 1);
+ options.set(CdcRecordStoreWriteOperator.SKIP_CORRUPT_RECORD, true);
+ FileStoreTable table =
+ createFileStoreTable(
+ rowTypeWithComplexColumn(
+ DataTypes.MAP(DataTypes.STRING(),
DataTypes.STRING())),
+ Collections.emptyList(),
+ Collections.singletonList("k"),
+ options);
+
+ OneInputStreamOperatorTestHarness<CdcRecord, Committable> harness =
+ createTestHarness(table);
+ harness.open();
+
+ harness.processElement(recordWithValue("{\"a\": "), 1);
+ harness.prepareSnapshotPreBarrier(1);
+
+ assertThat(harness.extractOutputValues())
+ .allSatisfy(
+ committable ->
+ assertThat(
+ ((CommitMessageImpl)
committable.commitMessage())
+ .newFilesIncrement()
+ .newFiles())
+ .isEmpty());
+
+ harness.close();
+ }
+
private OneInputStreamOperatorTestHarness<CdcRecord, Committable>
createTestHarness(
FileStoreTable table) throws Exception {
CdcRecordStoreWriteOperator.Factory operatorFactory =
@@ -279,9 +385,19 @@ public class CdcRecordStoreWriteOperatorTest {
private FileStoreTable createFileStoreTable(
RowType rowType, List<String> partitions, List<String>
primaryKeys) throws Exception {
+ return createFileStoreTable(rowType, partitions, primaryKeys, new
Options());
+ }
+
+ private FileStoreTable createFileStoreTable(
+ RowType rowType,
+ List<String> partitions,
+ List<String> primaryKeys,
+ Options extraOptions)
+ throws Exception {
Options conf = new Options();
conf.set(CdcRecordStoreWriteOperator.RETRY_SLEEP_TIME,
Duration.ofMillis(10));
conf.set(CoreOptions.BUCKET, 1);
+ extraOptions.toMap().forEach(conf::set);
TableSchema tableSchema =
SchemaUtils.forceCommit(
