[
https://issues.apache.org/jira/browse/FLINK-41272?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Seungki Kim updated FLINK-41272:
--------------------------------
Description:
*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}
was:
*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.
Compatibility of Debezium 1.9.8 with Kafka Connect 3.6 needs to be verified.
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.
*Workaround*
A custom DebeziumDeserializationSchema that rebuilds the before/after schemas
without
defaults and copies values with Struct#getWithoutDefault before calling
JsonConverter.
{code:java}
{code}
> [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)
