[
https://issues.apache.org/jira/browse/FLINK-41272?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18134295#comment-18134295
]
Seungki Kim commented on FLINK-41272:
-------------------------------------
My team and I ran into this in our production MySQL CDC pipeline, where binlog
events were silently turning NULLs into column defaults.
KIP-581 added `replace.null.with.default` to JsonConverter in Kafka 3.5, but
the Kafka Connect bundled with Flink CDC is 3.2.0 (pulled in by Debezium
1.9.8). I'd like to propose upgrading the
bundled Kafka Connect to 3.6.0. JsonDebeziumDeserializationSchema already
passes customConverterConfigs to JsonConverter, so users could then set
`replace.null.with.default=false` without any new API, while the default (true)
keeps existing behavior unchanged.
Since Debezium 1.9.8 was built against Kafka 3.2, compatibility with Kafka
Connect 3.6 needs to be verified. I'm happy to take care of that as part of the
contribution. If the upgrade turns out not to be feasible, an alternative is to
add an equivalent option in Flink CDC that converts the "before"/"after" images
with a copy of the schema whose field defaults are removed.
I'm open to other approaches if you think there's a better way.
Once we agree on the approach, could a committer assign this issue to me?
> [mysql-cdc] NULL values of nullable columns with a DEFAULT are emitted as the
> default value in Debezium JSON
> ------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41272
> URL: https://issues.apache.org/jira/browse/FLINK-41272
> Project: Flink
> Issue Type: Bug
> Components: Flink CDC
> Affects Versions: cdc-3.6.0
> Environment: * flink-sql-connector-mysql-cdc 3.6.0-2.2 (Debezium
> 1.9.8.Final, bundled Kafka Connect 3.2.0, shaded)
> * MySQL 8.0 (Aurora MySQL 3)
> * JsonConverter configured with schemas.enable=false
> Reporter: Seungki Kim
> Priority: Major
>
> *Problem*
> For a MySQL column that is nullable and has a DEFAULT (e.g. `count INT NULL
> DEFAULT 0`),
> a row whose stored value is NULL is emitted as the column default (0) instead
> of null
> when change events are converted to JSON with
> JsonDebeziumDeserializationSchema
> (or the shaded Kafka Connect JsonConverter directly).
> Snapshot reads of the same rows from the database return NULL, so the binlog
> output
> silently diverges from the source.
> This was previously reported on GitHub (ververica/flink-cdc-connectors#516)
> and closed
> because it predated 2.3.0. It still reproduces on 3.6.0.
>
> *Steps to reproduce*
> {code:java}
> CREATE TABLE t (id BIGINT PRIMARY KEY, cnt INT NULL DEFAULT 0, note
> VARCHAR(10));
> INSERT INTO t (id, cnt, note) VALUES (1, NULL, 'a');
> UPDATE t SET note = 'b' WHERE id = 1;{code}
> Expected "after": \{"id":1,"cnt":null,"note":"b"}
> Actual "after": \{"id":1,"cnt":0,"note":"b"}
>
>
> *Root cause*
> Debezium copies the column DEFAULT into the Connect field schema
> (defaultValue).
> Kafka Connect's JsonConverter writes schema.defaultValue() whenever the field
> value is null.
> KIP-581 added `replace.null.with.default` (Kafka 3.5) to disable this, but
> the Kafka Connect
> version shaded into flink-sql-connector-mysql-cdc is 3.2.0, so
> JsonConverterConfig only
> supports schemas.enable, schemas.cache.size and decimal.format. Passing
> replace.null.with.default via customConverterConfigs is silently ignored.
> Because the classes are relocated, users cannot swap in a newer connect-json
> either.
>
> *Impact*
> Every nullable column with a non-null DEFAULT is affected. Downstream tables
> built from the
> binlog get default values where the source has NULL. In our case this
> affected 87 columns
> across 11 tables, including DATETIME columns defaulting to '1970-01-01
> 00:00:00'.
>
> *Possible fixes*
> Expose an equivalent of Kafka's `replace.null.with.default` (KIP-581) in
> Flink CDC,
> e.g. as a JsonDebeziumDeserializationSchema / connector option, defaulting to
> true so that
> existing output does not change. Setting it to false would keep NULL as null.
> It could be implemented either by:
> # Bundling Kafka Connect >= 3.5 and passing the option through to
> JsonConverter, or
> # Without a Kafka upgrade, converting the row images ("before"/"after") with
> a schema copy whose field defaults are removed, reading values via
> Struct#getWithoutDefault.
>
> *Workaround*
> A custom DebeziumDeserializationSchema that rebuilds the before/after schemas
> without
> defaults and copies values with Struct#getWithoutDefault before calling
> JsonConverter.
> {code}
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
