[
https://issues.apache.org/jira/browse/FLINK-41272?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18134310#comment-18134310
]
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.
To fix it, I rebuilt flink-sql-connector-mysql-cdc 3.6.0-2.2 with Kafka Connect
3.6.0 (same artifacts and relocations) and set
`replace.null.with.default=false` via the converter configs. The verification
details are in the description, and we're and have been running it in
production to our pipeline.
I'd like to contribute this upstream, upgrade the bundled Kafka Connect to
3.6.0, implement OffsetBackingStore#connectorPartitions in
FlinkOffsetBackingStore.
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*
> *Option 1.* Upgrade the bundled Kafka Connect to 3.6.0 (currently 3.2.0,
> pulled in by Debezium 1.9.8). JsonConverter then supports KIP-581's
> `replace.null.with.default`, and since JsonDebeziumDeserializationSchema
> already passes customConverterConfigs to JsonConverter, users can set it to
> false without any new API. The default (true) keeps existing output unchanged.
> Things to handle with the upgrade (see Verification below):
> * FlinkOffsetBackingStore should implement
> OffsetBackingStore#connectorPartitions(String), added in Kafka 3.5 .(KIP-875)
> * Keep the bundled Jackson on a single version; Kafka Connect 3.6 pulls in
> 2.13.5.
> *Option 2.* If the upgrade is not feasible, expose an equivalent option in
> Flink CDC (default true) that converts the row images ("before"/"after") with
> a schema copy whose field defaults are removed, reading values via
> Struct#getWithoutDefault.
>
> *Verification*
> I rebuilt flink-sql-connector-mysql-cdc 3.6.0-2.2 with the same artifact set
> and relocations, replacing only Kafka Connect 3.2.0 with 3.6.0, and checked:
> * All 363 Kafka classes/methods/fields referenced by Flink CDC and Debezium
> 1.9.8 resolve in 3.6.0.
> * The only new abstract method on a Kafka type implemented by Flink
> CDC/Debezium is OffsetBackingStore#connectorPartitions(String), missing in
> FlinkOffsetBackingStore. It is only called by the Connect Worker /
> ConnectorOffsetBackingStore (offsets REST API), which the embedded engine
> does not use, so it is not hit at runtime.
> * Mixing Jackson versions inside the bundle breaks JsonConverter
> (NoClassDefFoundError for StreamConstraintsException); with a single Jackson
> version (2.13.2, as in the 3.6.0-2.2 bundle) it works.
> * With `replace.null.with.default=false`, snapshot and binlog events from
> MySQL 8.0 keep NULL in a `INT NULL DEFAULT 0` column, while omitted columns
> still get the default.
>
> *Workaround*
> A custom DebeziumDeserializationSchema that rebuilds the before/after schemas
> without
> defaults and copies values with Struct#getWithoutDefault before calling
> JsonConverter.
> {code:java}
> {code}
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
