From 7d7ba2dc19e95728719ef0dc64a53fe7329bc10c Mon Sep 17 00:00:00 2001
From: Kunni <lvyanquan.lyq@alibaba-inc.com>
Date: Tue, 16 Jan 2024 14:33:40 +0800
Subject: [PATCH] [hotfix][cdc-connector][mysql] Skip SchemaChangeEvents that
 were not included in capturedTableFilter

---
 .../connectors/mysql/source/MySqlPipelineITCase.java |  6 ++++++
 .../mysql/debezium/reader/BinlogSplitReader.java     | 12 ++++++++++--
 .../connectors/mysql/source/utils/RecordUtils.java   |  8 ++++++++
 3 files changed, 24 insertions(+), 2 deletions(-)

diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/com/ververica/cdc/connectors/mysql/source/MySqlPipelineITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/com/ververica/cdc/connectors/mysql/source/MySqlPipelineITCase.java
index 0e4b78358..00bdda4ba 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/com/ververica/cdc/connectors/mysql/source/MySqlPipelineITCase.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/com/ververica/cdc/connectors/mysql/source/MySqlPipelineITCase.java
@@ -522,6 +522,12 @@ public class MySqlPipelineITCase extends MySqlSourceTestBase {
                         tableId,
                         Collections.singletonList(
                                 Column.physicalColumn("DESC3", DataTypes.BIGINT()))));
+
+        // Should not catch SchemaChangeEvent of tables other than `products`
+        statement.execute(
+                String.format(
+                        "ALTER TABLE `%s`.`orders` ADD COLUMN `desc1` VARCHAR(45) NULL;",
+                        inventoryDatabase.getDatabaseName()));
         return expected;
     }
 }
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java
index f8fc488fe..a9add78df 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java
@@ -65,6 +65,8 @@ import static com.ververica.cdc.connectors.mysql.source.utils.RecordUtils.getSpl
 import static com.ververica.cdc.connectors.mysql.source.utils.RecordUtils.getStructContainsChunkKey;
 import static com.ververica.cdc.connectors.mysql.source.utils.RecordUtils.getTableId;
 import static com.ververica.cdc.connectors.mysql.source.utils.RecordUtils.isDataChangeRecord;
+import static com.ververica.cdc.connectors.mysql.source.utils.RecordUtils.isSchemaChangeEvent;
+import static com.ververica.cdc.connectors.mysql.source.utils.RecordUtils.isTableChangeRecord;
 
 /**
  * A Debezium binlog reader implementation that also support reads binlog and filter overlapping
@@ -252,9 +254,15 @@ public class BinlogSplitReader implements DebeziumReader<SourceRecords, MySqlSpl
             }
             // not in the monitored splits scope, do not emit
             return false;
+        } else if (isSchemaChangeEvent(sourceRecord)) {
+            if (isTableChangeRecord(sourceRecord)) {
+                TableId tableId = getTableId(sourceRecord);
+                return capturedTableFilter.isIncluded(tableId);
+            } else {
+                // Not related to changes in table structure, like `CREATE/DROP DATABASE`, skip it
+                return false;
+            }
         }
-        // always send the schema change event and signal event
-        // we need record them to state of Flink
         return true;
     }
 
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/source/utils/RecordUtils.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/source/utils/RecordUtils.java
index e1cb52055..8423c6a64 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/source/utils/RecordUtils.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/source/utils/RecordUtils.java
@@ -18,6 +18,7 @@ package com.ververica.cdc.connectors.mysql.source.utils;
 
 import org.apache.flink.table.types.logical.RowType;
 
+import com.ververica.cdc.common.utils.StringUtils;
 import com.ververica.cdc.connectors.mysql.debezium.dispatcher.SignalEventDispatcher.WatermarkKind;
 import com.ververica.cdc.connectors.mysql.debezium.reader.DebeziumReader;
 import com.ververica.cdc.connectors.mysql.source.offset.BinlogOffset;
@@ -387,6 +388,13 @@ public class RecordUtils {
         return new TableId(dbName, null, tableName);
     }
 
+    public static boolean isTableChangeRecord(SourceRecord dataRecord) {
+        Struct value = (Struct) dataRecord.value();
+        Struct source = value.getStruct(Envelope.FieldName.SOURCE);
+        String tableName = source.getString(TABLE_NAME_KEY);
+        return !StringUtils.isNullOrWhitespaceOnly(tableName);
+    }
+
     public static Object[] getSplitKey(
             RowType splitBoundaryType, SchemaNameAdjuster nameAdjuster, Struct target) {
         // the split key field contains single field now