[FLINK-36549]Fix using the ignore-parse-errors parameter in Debezium/Canal/Maxwell/Ogg JSON results in unexpected data loss #25919
Conversation
|
@flinkbot run azure |
… Debezium/Canal/Maxwell/Ogg JSON results in unexpected data loss
xiaoyu19950911
force-pushed
the
FLINK-36549
branch
from
5f67174 to
bd6e112
Compare
|
Reviewed by Chi on 09/01/2025 Need a committer to review |
There was a problem hiding this comment.
@xiaoyu19950911 Thank you for your contribution, I've left my comments, the PR is generally in a good shape.
| @@ -214,6 +215,7 @@ public void deserialize(@Nullable byte[] message, Collector<RowData> out) throws | |||
| if (message == null || message.length == 0) { | |||
| return; | |||
| } | |||
| List<GenericRowData> genericRowDataList = new ArrayList<>(); | |||
There was a problem hiding this comment.
We can add a transient field to CacalJsonDeserializationSchema to avoid creating new list for every record, it's more efficient, WDYT?
| @@ -73,10 +76,69 @@ void testFilteringTables() throws Exception { | |||
| InternalTypeInfo.of(PHYSICAL_DATA_TYPE.getLogicalType())) | |||
| .setDatabase("^my.*") | |||
| .setTable("^prod.*") | |||
| .setIgnoreParseErrors(true) | |||
There was a problem hiding this comment.
What's the purpose of enabling ignore-parse-error for this test case?
|
|
||
| @Override | ||
| public void collect(RowData record) { | ||
| if (record.getRowKind().equals(RowKind.DELETE) && record.getInt(0) == 103) { |
There was a problem hiding this comment.
The criteria is too specific, could we just throw exception directly no matter what the data is? We can only test the case that when exception occurs in chained downstream operator, the exception is correctly thrown.
| @@ -319,4 +381,22 @@ public void close() { | |||
| // do nothing | |||
| } | |||
| } | |||
|
|
|||
| private static class SpecialCollector implements Collector<RowData> { | |||
There was a problem hiding this comment.
How about naming it ThrowingExceptionCollector?
What is the purpose of the change
Enabling the ignore-parse-errors parameter does not skip exceptions caused by the format of the emitted data.
Brief change log
Verifying this change
This change added tests and can be verified as follows:
CanalJsonSerDeSchemaTest#testIgnoreParseErrorsDebeziumJsonSerDeSchemaTest#testIgnoreParseErrorsMaxwellJsonSerDerTest#testIgnoreParseErrorsOggJsonSerDeSchemaTest#testIgnoreParseErrorsDoes this pull request potentially affect one of the following parts:
@Public(Evolving): (no)Documentation