dev-donghwan opened a new pull request, #4610: URL: https://github.com/apache/flink-cdc/pull/4610
## What is the purpose of this pull request? [FLINK-41317](https://issues.apache.org/jira/browse/FLINK-41317) The Kafka sink's `canal-json` and `debezium-json` formats throw a `NullPointerException` on `REPLACE` events. They handle `UPDATE` and `REPLACE` in the same branch and read `before()` unconditionally, while `REPLACE` carries only the after image. This is reachable today with the Fluss pipeline source (since 3.7.0), which emits `UPDATE_AFTER` as `REPLACE`: updating a row of a Fluss primary-key table makes a Fluss -> Kafka pipeline fail. It will also be reachable with the PostgreSQL upsert changelog mode proposed in #4437. Same class of issue as FLINK-39684 for the Paimon sink. ## Brief change log - `DebeziumJsonSerializationSchema`: write `"before": null` when the event has no before image. - `CanalJsonSerializationSchema`: write `"old": null` when the event has no before image. Note on the output format: `{"before": null, "op": "u"}` matches what Debezium itself emits for PostgreSQL tables with `REPLICA IDENTITY DEFAULT`. Flink's own `debezium-json` / `canal-json` decoders reject an update without a before image, so downstream Flink SQL jobs reading these topics still need one. An alternative would be to emit `REPLACE` as an insert (`c` / `INSERT`), which upsert consumers handle by key. I kept the faithful form here; happy to switch if that is preferred. --- ## Verifying this change This change added tests and can be verified as follows: - Added `REPLACE` cases to `DebeziumJsonSerializationSchemaTest` and `CanalJsonSerializationSchemaTest`. They fail on master with the NPE and pass with this change. - All Kafka serializer tests pass (17/17). - Manually verified with `FlussSourcePipelineITCase` that a Fluss primary-key table update is emitted as `REPLACE` with `before == null` (not included in this PR). ## Documentation - Does this pull request introduce a new feature? no - If yes, how is the feature documented? not applicable --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) Generated-by: Claude Code (Claude Opus 5.5) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
