[
https://issues.apache.org/jira/browse/FLINK-41317?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Yanquan Lv resolved FLINK-41317.
--------------------------------
Fix Version/s: cdc-3.8.0
Assignee: kim
Resolution: Fixed
Fixed in master via f830a687715fff53a6e3501c23904bdc227bdd32.
> [pipeline-connector/kafka] NullPointerException when serializing REPLACE
> events in canal-json and debezium-json
> ---------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41317
> URL: https://issues.apache.org/jira/browse/FLINK-41317
> Project: Flink
> Issue Type: Bug
> Components: Flink CDC
> Affects Versions: cdc-3.7.0
> Reporter: kim
> Assignee: kim
> Priority: Major
> Labels: pull-request-available
> Fix For: cdc-3.8.0
>
>
> The Kafka pipeline sink fails with a NullPointerException when it receives a
> {{REPLACE}} event.
> {{CanalJsonSerializationSchema}} and {{DebeziumJsonSerializationSchema}}
> handle {{UPDATE}} and {{REPLACE}} in the same branch and pass {{before()}} to
> {{TableSchemaInfo#getRowDataFromRecordData}} unconditionally. {{REPLACE}}
> events carry only the after image, so {{before()}} is null:
> {code}
> Caused by: java.lang.NullPointerException
> at
> org.apache.flink.cdc.connectors.kafka.json.TableSchemaInfo.getRowDataFromRecordData(TableSchemaInfo.java:83)
> at
> org.apache.flink.cdc.connectors.kafka.json.debezium.DebeziumJsonSerializationSchema.convertUpdateEventToRowData(DebeziumJsonSerializationSchema.java:465)
> {code}
> *How to reproduce*
> The Fluss pipeline source (since cdc-3.7.0) emits {{UPDATE_AFTER}} as
> {{REPLACE}} and drops {{UPDATE_BEFORE}}. Updating a row of a Fluss
> primary-key table (e.g. inserting the same key again) produces {{+I[1,
> Alice]}} followed by {{+R[1, Alicia]}} with {{before == null}}. A Fluss ->
> Kafka pipeline with {{value.format}} set to {{debezium-json}} or
> {{canal-json}} then fails on the first update.
> The same applies to any other source that emits {{REPLACE}}, such as the
> proposed PostgreSQL {{changelog-mode: upsert}}. The Paimon sink had the same
> issue and has already been fixed.
> *Proposed fix*
> Write the before image as null when it is absent: {{"before": null}} for
> debezium-json and {{"old": null}} for canal-json.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
