Seungki Kim created FLINK-41272:
-----------------------------------
Summary: [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
*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)
