[
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: to be confirmed for this run (previous report: 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
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.
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 sends
a DropTableEvent for a table listed in tables.exclude to SchemaCoordinator.
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'
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:none}
Both DROP records have binlog.000372, pos=492809. SchemaCoordinator receives
DropTableEvent{tableId=cdc_filter_repro.excluded_multi} and skips it as
redundant because its schema is null. Schema-change handling is LENIENT.
{code}
h3. Expected result
The Source suppresses schema-change events for excluded tables. The subscribed
table's DROP remains eligible for downstream schema handling.
h3. Key logs
{code:none}
2026-10-26 12:12:36,947 INFO
org.apache.flink.cdc.connectors.mysql.source.MySqlEventDeserializer [] - TRACE
DDL DROP TABLE `cdc_filter_repro`.`keep_multi` with position
file=binlog.000372,pos=492809 and gtids=<REDACTED>
2026-10-26 12:12:36,989 INFO
org.apache.flink.cdc.connectors.mysql.source.MySqlEventDeserializer [] - TRACE
DDL DROP TABLE `cdc_filter_repro`.`excluded_multi` with position
file=binlog.000372,pos=492809 and gtids=<REDACTED>
2026-10-26 12:12:36,987 INFO
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator [] -
Step 4 - After being normalized with LENIENT behavior, final schema change
events are: []
2026-10-26 12:12:37,086 INFO
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator [] -
Schema change event DropTableEvent{tableId=cdc_filter_repro.excluded_multi} is
redundant for current schema null, just skip it.
SchemaChangeRequest{tableId=cdc_filter_repro.excluded_multi,
schemaChangeEvent=DropTableEvent{tableId=cdc_filter_repro.excluded_multi},
subTaskId=0}
{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: to be confirmed for this run (previous report: 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
> 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.
> 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)
