[
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 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. 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
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. 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}
> 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 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. 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)
