[
https://issues.apache.org/jira/browse/FLINK-41272?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Seungki Kim updated FLINK-41272:
--------------------------------
Environment:
* flink-sql-connector-mysql-cdc 3.6.0-2.2 (Debezium 1.9.8.Final, bundled Kafka
Connect 3.2.0, shaded). master (3.8-SNAPSHOT) bundles the same Debezium and
Kafka Connect versions.
* Flink 2.2.1, Java 17
* MySQL 8.0 (Aurora MySQL 3)
* JsonConverter configured with schemas.enable=false (the null-to-default
replacement happens regardless of schemas.enable)
was:
* flink-sql-connector-mysql-cdc 3.6.0-2.2 (Debezium 1.9.8.Final, bundled Kafka
Connect 3.2.0, shaded). master (3.8-SNAPSHOT) bundles the same Debezium and
Kafka Connect versions.
* MySQL 8.0 (Aurora MySQL 3)
* JsonConverter configured with schemas.enable=false (the null-to-default
replacement happens regardless of schemas.enable)
> [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). master (3.8-SNAPSHOT)
> bundles the same Debezium and Kafka Connect versions.
> * Flink 2.2.1, Java 17
> * MySQL 8.0 (Aurora MySQL 3)
> * JsonConverter configured with schemas.enable=false (the null-to-default
> replacement happens regardless of schemas.enable)
> 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)
