[
https://issues.apache.org/jira/browse/FLINK-41197?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
丁笠 updated FLINK-41197:
-----------------------
Description:
h3. Summary
With scan.binlog.newly-added-table.enabled=true, a mixed multi-table DROP emits
an event for a table listed in tables.exclude. When Transform also matches that
table, the leaked event causes a Transform NPE and task failure.
h3. Environment
* Flink CDC: 3.4
* Flink: 1.20
* MySQL: 5.7
The exact CDC connector/runtime build and source commit for the reported run
are not yet identified. Standalone reproduction with an unmodified Apache
release has not been independently verified.
Standalone prerequisites: MySQL binlog enabled, binlog_format=ROW,
binlog_row_image=FULL, and an unused server-id range.
h3. Configuration
{code:yaml}
source:
type: mysql
hostname: <MYSQL_HOST>
port: 3306
username: <MYSQL_USER>
password: <REDACTED>
tables: 'cdc_filter_repro.\.*'
tables.exclude:
'cdc_filter_repro.excluded_single,cdc_filter_repro.excluded_multi'
server-id: 5500-5504
scan.binlog.newly-added-table.enabled: 'true'
schema-change.enabled: 'true'
transform:
- source-table: 'cdc_filter_repro.\.*'
projection: '\*'
sink:
type: values
print.enabled: true
pipeline:
parallelism: 1
{code}
For standalone reproduction, load MySQL and Values pipeline connector JARs
matching the chosen CDC version. Values is a repository test connector and is
not bundled in the default distribution.
h3. Reproduction
Use a fresh test database named cdc_filter_repro.
1. Before starting the job:
{code:sql}
CREATE DATABASE IF NOT EXISTS cdc_filter_repro;
USE cdc_filter_repro;
SHOW VARIABLES LIKE 'binlog_format';
SHOW VARIABLES LIKE 'binlog_row_image';
CREATE TABLE seed (id INT PRIMARY KEY, name VARCHAR(32));
INSERT INTO seed VALUES (1, 'seed');
{code}
2. Start the job and wait for seed snapshot completion and binlog streaming.
Then execute:
{code:sql}
CREATE TABLE excluded_single (id INT PRIMARY KEY);
INSERT INTO excluded_single VALUES (1);
DROP TABLE excluded_single;
CREATE TABLE keep_multi (id INT PRIMARY KEY);
CREATE TABLE excluded_multi (id INT PRIMARY KEY);
INSERT INTO keep_multi VALUES (1);
INSERT INTO excluded_multi VALUES (1);
{code}
3. Wait for keep_multi CREATE/INSERT output, then execute:
{code:sql}
DROP TABLE keep_multi, excluded_multi;
{code}
h3. Actual result
{code}
The two DROP records share binlog.000374, pos=751768.
DropTableEvent{tableId=cdc_filter_repro.excluded_multi} reaches
PostTransformOperator and triggers TransformException caused by
NullPointerException at processSchemaChangeEvent
(PostTransformOperator.java:240). The Transform task switches from RUNNING to
FAILED.
{code}
The earlier run without Transform also emitted the excluded table's DROP to
SchemaCoordinator.
h3. Expected result
The Source filters the excluded table's DROP and retains the subscribed table's
event. An excluded table's DROP must not enter Transform or fail the job.
Preventing the Transform NPE alone does not fix the Source filtering issue.
h3. Key logs
{code}
2026-10-26 14:25:30,804 INFO
org.apache.flink.cdc.connectors.mysql.source.MySqlEventDeserializer [] - TRACE
DDL DROP TABLE `cdc_filter_repro`.`keep_multi` with position
file=binlog.000374,pos=751768 and gtids=<REDACTED>
2026-10-26 14:25:30,808 INFO
org.apache.flink.cdc.connectors.mysql.source.MySqlEventDeserializer [] - TRACE
DDL DROP TABLE `cdc_filter_repro`.`excluded_multi` with position
file=binlog.000374,pos=751768 and gtids=<REDACTED>
2026-10-26 14:25:30,934 INFO
org.apache.flink.runtime.executiongraph.ExecutionGraph [] -
Transform:Data -> SchemaOperator -> PrePartition (1/1) <REDACTED_TASK> switched
from RUNNING to FAILED.
org.apache.flink.cdc.runtime.operators.transform.exceptions.TransformException:
Failed to post-transform with
DropTableEvent{tableId=cdc_filter_repro.excluded_multi}
at
org.apache.flink.cdc.runtime.operators.transform.PostTransformOperator.processElement(PostTransformOperator.java:146)
...
Caused by: java.lang.NullPointerException
at
org.apache.flink.cdc.common.utils.Preconditions.checkNotNull(Preconditions.java:44)
at
org.apache.flink.cdc.runtime.operators.transform.PostTransformOperator.processSchemaChangeEvent(PostTransformOperator.java:240)
at
org.apache.flink.cdc.runtime.operators.transform.PostTransformOperator.processElementInternal(PostTransformOperator.java:178)
at
org.apache.flink.cdc.runtime.operators.transform.PostTransformOperator.processElement(PostTransformOperator.java:130)
{code}
was:
h3. Summary
With scan.binlog.newly-added-table.enabled=true, a mixed multi-table DROP emits
an event for a table listed in tables.exclude. When Transform also matches that
table, the leaked event causes a Transform NPE and task failure.
h3. Environment
* Flink CDC: 3.4
* Flink: 1.20
* MySQL: 5.7
h3. Configuration
{code:yaml}
source:
type: mysql
hostname: <MYSQL_HOST>
port: 3306
username: <MYSQL_USER>
password: <REDACTED>
tables: 'cdc_filter_repro.\.*'
tables.exclude:
'cdc_filter_repro.excluded_single,cdc_filter_repro.excluded_multi'
server-id: 5500-5504
scan.binlog.newly-added-table.enabled: 'true'
schema-change.enabled: 'true'
transform:
- source-table: 'cdc_filter_repro.\.*'
projection: '\*'
sink:
type: values
print.enabled: true
pipeline:
parallelism: 1
{code}
h3. Reproduction
Database: cdc_filter_repro.
1. Before starting the job:
{code:sql}
CREATE TABLE seed (id INT PRIMARY KEY, name VARCHAR(32));
INSERT INTO seed VALUES (1, 'seed');
{code}
2. Start the job and wait for seed snapshot completion and binlog streaming.
Then execute:
{code:sql}
CREATE TABLE excluded_single (id INT PRIMARY KEY);
INSERT INTO excluded_single VALUES (1);
DROP TABLE excluded_single;
CREATE TABLE keep_multi (id INT PRIMARY KEY);
CREATE TABLE excluded_multi (id INT PRIMARY KEY);
INSERT INTO keep_multi VALUES (1);
INSERT INTO excluded_multi VALUES (1);
{code}
3. Wait for keep_multi CREATE/INSERT output, then execute:
{code:sql}
DROP TABLE keep_multi, excluded_multi;
{code}
h3. Actual result
{code}
The two DROP records share binlog.000374, pos=751768.
DropTableEvent{tableId=cdc_filter_repro.excluded_multi} reaches
PostTransformOperator and triggers TransformException caused by
NullPointerException at processSchemaChangeEvent
(PostTransformOperator.java:240). The Transform task switches from RUNNING to
FAILED.
{code}
The earlier run without Transform also emitted the excluded table's DROP to
SchemaCoordinator.
h3. Expected result
The Source filters the excluded table's DROP and retains the subscribed table's
event. An excluded table's DROP must not enter Transform or fail the job.
h3. Key logs
{code}
2026-10-26 14:25:30,804 INFO
org.apache.flink.cdc.connectors.mysql.source.MySqlEventDeserializer [] - TRACE
DDL DROP TABLE `cdc_filter_repro`.`keep_multi` with position
file=binlog.000374,pos=751768 and gtids=<REDACTED>
2026-10-26 14:25:30,808 INFO
org.apache.flink.cdc.connectors.mysql.source.MySqlEventDeserializer [] - TRACE
DDL DROP TABLE `cdc_filter_repro`.`excluded_multi` with position
file=binlog.000374,pos=751768 and gtids=<REDACTED>
2026-10-26 14:25:30,934 INFO
org.apache.flink.runtime.executiongraph.ExecutionGraph [] -
Transform:Data -> SchemaOperator -> PrePartition (1/1) <REDACTED_TASK> switched
from RUNNING to FAILED.
org.apache.flink.cdc.runtime.operators.transform.exceptions.TransformException:
Failed to post-transform with
DropTableEvent{tableId=cdc_filter_repro.excluded_multi}
at
org.apache.flink.cdc.runtime.operators.transform.PostTransformOperator.processElement(PostTransformOperator.java:146)
...
Caused by: java.lang.NullPointerException
at
org.apache.flink.cdc.common.utils.Preconditions.checkNotNull(Preconditions.java:44)
at
org.apache.flink.cdc.runtime.operators.transform.PostTransformOperator.processSchemaChangeEvent(PostTransformOperator.java:240)
at
org.apache.flink.cdc.runtime.operators.transform.PostTransformOperator.processElementInternal(PostTransformOperator.java:178)
at
org.apache.flink.cdc.runtime.operators.transform.PostTransformOperator.processElement(PostTransformOperator.java:130)
{code}
> Multi-table DROP may bypass tables.exclude when binlog newly-added table
> discovery is enabled
> ---------------------------------------------------------------------------------------------
>
> Key: FLINK-41197
> URL: https://issues.apache.org/jira/browse/FLINK-41197
> Project: Flink
> Issue Type: Bug
> Components: Flink CDC
> Reporter: 丁笠
> Priority: Major
>
> h3. Summary
> With scan.binlog.newly-added-table.enabled=true, a mixed multi-table DROP
> emits an event for a table listed in tables.exclude. When Transform also
> matches that table, the leaked event causes a Transform NPE and task failure.
> h3. Environment
> * Flink CDC: 3.4
> * Flink: 1.20
> * MySQL: 5.7
> The exact CDC connector/runtime build and source commit for the reported run
> are not yet identified. Standalone reproduction with an unmodified Apache
> release has not been independently verified.
> Standalone prerequisites: MySQL binlog enabled, binlog_format=ROW,
> binlog_row_image=FULL, and an unused server-id range.
> h3. Configuration
> {code:yaml}
> source:
> type: mysql
> hostname: <MYSQL_HOST>
> port: 3306
> username: <MYSQL_USER>
> password: <REDACTED>
> tables: 'cdc_filter_repro.\.*'
> tables.exclude:
> 'cdc_filter_repro.excluded_single,cdc_filter_repro.excluded_multi'
> server-id: 5500-5504
> scan.binlog.newly-added-table.enabled: 'true'
> schema-change.enabled: 'true'
> transform:
> - source-table: 'cdc_filter_repro.\.*'
> projection: '\*'
> sink:
> type: values
> print.enabled: true
> pipeline:
> parallelism: 1
> {code}
> For standalone reproduction, load MySQL and Values pipeline connector JARs
> matching the chosen CDC version. Values is a repository test connector and is
> not bundled in the default distribution.
> h3. Reproduction
> Use a fresh test database named cdc_filter_repro.
> 1. Before starting the job:
> {code:sql}
> CREATE DATABASE IF NOT EXISTS cdc_filter_repro;
> USE cdc_filter_repro;
> SHOW VARIABLES LIKE 'binlog_format';
> SHOW VARIABLES LIKE 'binlog_row_image';
> CREATE TABLE seed (id INT PRIMARY KEY, name VARCHAR(32));
> INSERT INTO seed VALUES (1, 'seed');
> {code}
> 2. Start the job and wait for seed snapshot completion and binlog streaming.
> Then execute:
> {code:sql}
> CREATE TABLE excluded_single (id INT PRIMARY KEY);
> INSERT INTO excluded_single VALUES (1);
> DROP TABLE excluded_single;
> CREATE TABLE keep_multi (id INT PRIMARY KEY);
> CREATE TABLE excluded_multi (id INT PRIMARY KEY);
> INSERT INTO keep_multi VALUES (1);
> INSERT INTO excluded_multi VALUES (1);
> {code}
> 3. Wait for keep_multi CREATE/INSERT output, then execute:
> {code:sql}
> DROP TABLE keep_multi, excluded_multi;
> {code}
> h3. Actual result
> {code}
> The two DROP records share binlog.000374, pos=751768.
> DropTableEvent{tableId=cdc_filter_repro.excluded_multi} reaches
> PostTransformOperator and triggers TransformException caused by
> NullPointerException at processSchemaChangeEvent
> (PostTransformOperator.java:240). The Transform task switches from RUNNING to
> FAILED.
> {code}
> The earlier run without Transform also emitted the excluded table's DROP to
> SchemaCoordinator.
> h3. Expected result
> The Source filters the excluded table's DROP and retains the subscribed
> table's event. An excluded table's DROP must not enter Transform or fail the
> job.
> Preventing the Transform NPE alone does not fix the Source filtering issue.
> h3. Key logs
> {code}
> 2026-10-26 14:25:30,804 INFO
> org.apache.flink.cdc.connectors.mysql.source.MySqlEventDeserializer [] -
> TRACE DDL DROP TABLE `cdc_filter_repro`.`keep_multi` with position
> file=binlog.000374,pos=751768 and gtids=<REDACTED>
> 2026-10-26 14:25:30,808 INFO
> org.apache.flink.cdc.connectors.mysql.source.MySqlEventDeserializer [] -
> TRACE DDL DROP TABLE `cdc_filter_repro`.`excluded_multi` with position
> file=binlog.000374,pos=751768 and gtids=<REDACTED>
> 2026-10-26 14:25:30,934 INFO
> org.apache.flink.runtime.executiongraph.ExecutionGraph [] -
> Transform:Data -> SchemaOperator -> PrePartition (1/1) <REDACTED_TASK>
> switched from RUNNING to FAILED.
> org.apache.flink.cdc.runtime.operators.transform.exceptions.TransformException:
> Failed to post-transform with
> DropTableEvent{tableId=cdc_filter_repro.excluded_multi}
> at
> org.apache.flink.cdc.runtime.operators.transform.PostTransformOperator.processElement(PostTransformOperator.java:146)
> ...
> Caused by: java.lang.NullPointerException
> at
> org.apache.flink.cdc.common.utils.Preconditions.checkNotNull(Preconditions.java:44)
> at
> org.apache.flink.cdc.runtime.operators.transform.PostTransformOperator.processSchemaChangeEvent(PostTransformOperator.java:240)
> at
> org.apache.flink.cdc.runtime.operators.transform.PostTransformOperator.processElementInternal(PostTransformOperator.java:178)
> at
> org.apache.flink.cdc.runtime.operators.transform.PostTransformOperator.processElement(PostTransformOperator.java:130)
> {code}
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
