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