[
https://issues.apache.org/jira/browse/FLINK-41197?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
丁笠 updated FLINK-41197:
-----------------------
Environment:
Flink CDC: 3.4，3.6
Flink: 1.20
MySQL: 5.7
was:
Flink CDC: 3.4
Flink: 1.20
MySQL: 5.7
> 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
> Affects Versions: cdc-3.4.0, cdc-3.6.0
> Environment: Flink CDC: 3.4，3.6
> Flink: 1.20
> MySQL: 5.7
> Reporter: 丁笠
> Priority: Major
> Fix For: cdc-3.6.0
>
>
> 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, 3.6.0
> * Flink: 1.20
> * MySQL: 5.7
> The initial run used an unidentified CDC build. For the follow-up, the
> reporter reports using the unmodified community MySQL pipeline connector
> described below, with the existing CDC 3.4 runtime and Flink 1.20 deployment.
> A complete unmodified Apache distribution has not been independently tested.
> 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. Follow-up with the community connector
> * Connector: flink-cdc-pipeline-connector-mysql-3.4.0.jar
> * Source tag: release-3.4.0
> * Source commit: 06581fc73b84b1da45fccf93441d96a118d9bfc1
> * Connector source: unmodified; compiled against Flink 1.20.1
> The follow-up JM log shows the same failure: at 16:00:01,
> PostTransformOperator receives
> DropTableEvent\{tableId=cdc_filter_repro.excluded_multi\} with unknown schema
> and throws NullPointerException at PostTransformOperator.java:240. The
> Transform task changes from RUNNING to FAILED and the job enters RESTARTING.
> The JM excerpt does not record the loaded connector JAR, so connector
> provenance is based on the delivered build and the reporter's confirmation.
> h3. Reproduction with CDC 3.6.0
> * MySQL connector: flink-cdc-pipeline-connector-mysql-3.6.0.jar
> * Framework: flink-cdc-common, flink-cdc-runtime and flink-cdc-flink1-compat
> 3.6.0
> * Flink: 1.20; MySQL: 5.7
> With the same filtering configuration and SQL, the job enters RUNNING at
> 17:14:51. At 17:15:29, PostTransformOperator receives
> DropTableEvent\{tableId=cdc_filter_repro.excluded_multi\} with unknown schema
> and throws NullPointerException at PostTransformOperator.java:240. The job
> then enters RESTARTING.
> 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)
