[
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.
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}
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*
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}
> [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. 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}
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
