kim created FLINK-41317:
---------------------------
Summary: [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
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)
