[
https://issues.apache.org/jira/browse/FLINK-41197?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18132121#comment-18132121
]
丁笠 commented on FLINK-41197:
----------------------------
h3. Follow-up: real-job reproduction and sanitized evidence
A real MySQL job has now been exercised using the configuration and SQL below.
The new evidence shows that a DropTableEvent for an explicitly excluded table
reaches SchemaCoordinator after a mixed multi-table DROP. This updates the
initial component-only validation status.
The exact CDC, Flink and MySQL versions have not been verified from these
excerpts. The configuration retains the submitted runtime's single-entry
sources wrapper. This is a real-job observation in that runtime, not yet a
verified reproduction on an unmodified upstream release.
h3. Submitted configuration (sanitized)
Display-only names have been removed; connection address, username and password
are placeholders. Relevant filter settings have been retained.
{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. SQL and execution sequence
The statements were executed in cdc_filter_repro. Start the job after creating
seed, wait for the initial snapshot and binlog streaming, execute the
single-table exclusion control, then create the two multi-table test tables.
Wait for keep_multi CREATE/INSERT to be consumed before issuing the single
mixed DROP statement.
{code:sql}
-- Execute in the cdc_filter_repro database.
-- Before starting the CDC job:
CREATE TABLE seed (id INT PRIMARY KEY, name VARCHAR(32));
INSERT INTO seed VALUES (1, 'seed');
-- Start the CDC job and wait for seed snapshot/data output and binlog
streaming.
-- Single-table exclusion control:
CREATE TABLE excluded_single (id INT PRIMARY KEY);
INSERT INTO excluded_single VALUES (1);
DROP TABLE excluded_single;
-- While the CDC job is running:
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);
-- Wait until keep_multi CREATE and INSERT have been consumed, then:
DROP TABLE keep_multi, excluded_multi;
{code}
h3. Key TaskManager evidence
The deserializer logs DROP for both keep_multi and excluded_multi with the same
binlog file/position (binlog.000372, pos=492809). GTID values are redacted. The
test-table identifiers and file/position are retained to correlate the two
events.
{code}
CreateTableEvent{tableId=cdc_filter_repro.seed, schema=columns={`id` INT NOT
NULL,`name` VARCHAR(32)}, primaryKeys=id, options=()}
DataChangeEvent{tableId=cdc_filter_repro.seed, before=[], after=[1, seed],
op=INSERT, meta=()}
CreateTableEvent{tableId=cdc_filter_repro.keep_multi, schema=columns={`id` INT
NOT NULL}, primaryKeys=id, options=()}
DataChangeEvent{tableId=cdc_filter_repro.keep_multi, before=[], after=[1],
op=INSERT, meta=()}
2026-10-26 12:12:06,469 INFO
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaOperator [] - 0>
Finished schema change events:
[CreateTableEvent{tableId=cdc_filter_repro.keep_multi, schema=columns={`id` INT
NOT NULL}, primaryKeys=id, options=()}]
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,988 INFO
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaOperator [] - 0>
Finished schema change events: []
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:37,088 INFO
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaOperator [] - 0>
Finished schema change events: []
2026-10-26 12:12:37,088 INFO
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaOperator [] - 0>
Refreshed evolved schemas: {cdc_filter_repro.excluded_multi=null}
{code}
h3. Key JobManager evidence
SchemaCoordinator receives the excluded table's DropTableEvent, sees its schema
as null and skips the event as redundant. The runtime logs show LENIENT
schema-change handling; the subscribed table's final evolved DROP event list is
also empty. Accordingly, this run demonstrates leakage to the schema
coordinator, rather than an excluded DROP printed by the Values Sink.
{code}
2026-10-26 12:10:38,110 INFO
org.apache.flink.cdc.connectors.mysql.utils.MySqlSchemaUtils [] - list
of available tables are: [cdc_filter_repro.seed]
2026-10-26 12:10:38,151 INFO
org.apache.flink.cdc.connectors.mysql.factory.MySqlDataSourceFactory [] - Final
Debezium-style table capture list: cdc_filter_repro\..*
2026-10-26 12:12:06,467 INFO
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator [] -
Step 4 - After being normalized with LENIENT behavior, final schema change
events are: [CreateTableEvent{tableId=cdc_filter_repro.keep_multi,
schema=columns={`id` INT NOT NULL}, primaryKeys=id, options=()}]
2026-10-26 12:12:36,986 INFO
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator [] -
Step 1 - Start deducing evolved schema change for
DropTableEvent{tableId=cdc_filter_repro.keep_multi}
2026-10-26 12:12:36,987 INFO
org.apache.flink.cdc.runtime.operators.schema.common.SchemaDerivator [] -
Ignored schema change DropTableEvent{tableId=cdc_filter_repro.keep_multi}.
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: []
SchemaChangeRequest{tableId=cdc_filter_repro.keep_multi,
schemaChangeEvent=DropTableEvent{tableId=cdc_filter_repro.keep_multi},
subTaskId=0}
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}
2026-10-26 12:15:01,855 INFO
org.apache.flink.cdc.runtime.operators.schema.common.SchemaRegistry [] -
Successfully completed checkpoint No.1
{code}
h3. Observed versus expected
Observed: DropTableEvent{tableId=cdc_filter_repro.excluded_multi} reaches
SchemaCoordinator despite the precise tables.exclude entry.
Expected: the excluded table's schema-change event should be suppressed by the
Source before it reaches downstream schema handling, while the subscribed
table's event remains eligible for that handling.
No relevant exception is shown in the supplied excerpts; a successful
checkpoint is recorded after the DROP. This update does not claim a Values Sink
failure or a Transform NPE.
h3. Remaining validation limits
* The single-table excluded_single control was executed, and no event for it
appears in the supplied excerpts. Absence in these excerpts is not claimed as
proof about the entire event stream.
* The excerpts do not expose the raw source.table value, the complete Debezium
table-change set or the representative table used for filtering. Therefore the
combined-table-name mechanism described in the initial component scenario
remains provisional. The real-job evidence establishes the excluded DROP
reaching downstream schema handling, without proving that precise mechanism.
* The observed LENIENT behavior was read from the runtime logs; no explicit
schema.change.behavior was added to the submitted YAML.
* Precise release versions and an upstream standalone reproduction remain to be
confirmed.
> 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
> When scan.binlog.newly-added-table.enabled is enabled, a MySQL schema-change
> record that contains multiple table names in a single source.table value may
> pass the subscription filter as one combined table identifier. Parsing the
> DROP statement afterwards can then emit a DropTableEvent for a table
> explicitly excluded by tables.exclude.
> This report is based on prior component-level observations with constructed
> schema-change records. A standalone public reproducer and validation against
> real MySQL binlogs are still being prepared. The exact affected upstream
> release(s), MySQL version and complete environment details will be provided
> after verification.
> h3. Triggering conditions
> * MySQL Pipeline Source has scan.binlog.newly-added-table.enabled set to true.
> * The inclusion pattern matches the combined table-name value.
> * tables.exclude precisely excludes one of the individual tables.
> * A multi-table DROP schema-change record represents the table names as a
> comma-separated source.table value.
> Relevant configuration fragment:
> {code:yaml}
> scan.binlog.newly-added-table.enabled: true
> tables.exclude: test_db.excluded_table
> {code}
> The inclusion pattern must be supplied with the reproducer. The fragment
> above is not a complete runnable pipeline.
> Illustrative SQL for the scenario to be validated:
> {code:sql}
> DROP TABLE test_db.captured_table, test_db.excluded_table;
> {code}
> Constructed metadata used in the component scenario:
> {code}
> source.db = test_db
> source.table = captured_table,excluded_table
> {code}
> The SQL above has not yet been verified to produce exactly this metadata in a
> real upstream MySQL/Debezium run.
> h3. Prior component-level observations
> The combined table-name value can match the inclusion pattern while failing
> to match the precise exclusion for test_db.excluded_table. The record is then
> accepted and the DDL parser emits individual DropTableEvent instances,
> including the excluded table's event.
> A single-table excluded DROP was filtered in the component control cases.
> Therefore this report is specifically about multi-table records under binlog
> newly-added table discovery, rather than all DROP events bypassing exclusion.
> h3. Expected behavior
> Apply inclusion and exclusion rules to each affected table. Emit the DROP
> event for the subscribed table and suppress the DROP event for the excluded
> table, without losing the subscribed table's event from the same statement.
> h3. Suspected root cause
> Record-level filtering uses a combined table identifier, while DDL parsing
> produces per-table events. The exclusion rules do not appear to be reapplied
> to each parsed event.
> h3. Reproducer and validation status
> A minimal public reproducer, full configuration, precise versions and raw
> event metadata will be added. Please treat the combined source.table value as
> a constructed component input until the real-binlog path is verified. We are
> not claiming end-to-end reproduction on all upstream releases.
> For the reproducer, the important assertions are that:
> * No DropTableEvent is emitted for the excluded table.
> * The DropTableEvent for the subscribed table is retained.
> * The single-table excluded DROP control remains filtered.
> h3. Related work
> FLINK-36945 and PR #3876 discuss related multi-table RENAME
> metadata/filtering behavior. They are referenced as related work; this report
> does not assume that the RENAME fix covers multi-table DROP.
> [FLINK-36945|https://issues.apache.org/jira/browse/FLINK-36945]
> [PR #3876|https://github.com/apache/flink-cdc/pull/3876]
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
