From 7822a1b9c0d0970d1bdecdede5b191c6f9189639 Mon Sep 17 00:00:00 2001
From: Ti Chi Robot <ti-community-prow-bot@tidb.io>
Date: Wed, 6 Dec 2023 14:20:51 +0800
Subject: [PATCH] ddl_manager (ticdc): fix in bdr mode cdc can not replicate a
 table's dmls after drop and re-create it  (#10080) (#10089)

close pingcap/tiflow#10079
---
 cdc/owner/ddl_manager.go                      | 27 ++++++++++++-------
 .../integration_tests/bdr_mode/data/start.sql |  1 +
 tests/integration_tests/bdr_mode/data/up.sql  |  5 ++++
 3 files changed, 24 insertions(+), 9 deletions(-)

diff --git a/cdc/owner/ddl_manager.go b/cdc/owner/ddl_manager.go
index 21fa767f41f..1943ffb604a 100644
--- a/cdc/owner/ddl_manager.go
+++ b/cdc/owner/ddl_manager.go
@@ -220,15 +220,6 @@ func (m *ddlManager) tick(
 			}
 
 			for _, event := range events {
-				// If changefeed is in BDRMode, skip ddl.
-				if m.BDRMode {
-					log.Info("changefeed is in BDRMode, skip a ddl event",
-						zap.String("namespace", m.changfeedID.Namespace),
-						zap.String("ID", m.changfeedID.ID),
-						zap.Any("ddlEvent", event))
-					continue
-				}
-
 				// TODO: find a better place to do this check
 				// check if the ddl event is belong to an ineligible table.
 				// If so, we should ignore it.
@@ -348,6 +339,24 @@ func (m *ddlManager) executeDDL(ctx context.Context) error {
 	if m.executingDDL == nil {
 		return nil
 	}
+
+	// If changefeed is in BDRMode, skip ddl.
+	if m.BDRMode {
+		log.Info("changefeed is in BDRMode, skip a ddl event",
+			zap.String("namespace", m.changfeedID.Namespace),
+			zap.String("ID", m.changfeedID.ID),
+			zap.Any("ddlEvent", m.executingDDL))
+		tableName := m.executingDDL.TableInfo.TableName
+		// Set it to nil first to accelerate GC.
+		m.pendingDDLs[tableName][0] = nil
+		m.pendingDDLs[tableName] = m.pendingDDLs[tableName][1:]
+		m.schema.DoGC(m.executingDDL.CommitTs - 1)
+		m.justSentDDL = m.executingDDL
+		m.executingDDL = nil
+		m.cleanCache()
+		return nil
+	}
+
 	failpoint.Inject("ExecuteNotDone", func() {
 		// This ddl will never finish executing.
 		// It is used to test the logic that a ddl only block the related table
diff --git a/tests/integration_tests/bdr_mode/data/start.sql b/tests/integration_tests/bdr_mode/data/start.sql
index 2d5007efbad..33e6cc75cba 100644
--- a/tests/integration_tests/bdr_mode/data/start.sql
+++ b/tests/integration_tests/bdr_mode/data/start.sql
@@ -3,3 +3,4 @@ create database `bdr_mode`;
 use `bdr_mode`;
 
 create table `t1` (id int primary key, name varchar(20));
+create table `t2` (id int primary key, name varchar(20));
\ No newline at end of file
diff --git a/tests/integration_tests/bdr_mode/data/up.sql b/tests/integration_tests/bdr_mode/data/up.sql
index dd926b15515..4e36df6157b 100644
--- a/tests/integration_tests/bdr_mode/data/up.sql
+++ b/tests/integration_tests/bdr_mode/data/up.sql
@@ -16,3 +16,8 @@ insert into `t1` values (22, '22'), (44, '44'), (66, '66'), (88, '88'), (108, '1
 rollback;
 
 insert into `t1` values (100, '100'), (300, '300'), (500, '500'), (700, '700'), (900, '900');
+
+drop table `t2`;
+create table `t2` (id int primary key, name varchar(20));
+insert into `t2` values (1, '1'), (3, '3'), (5, '5'), (7, '7'), (9, '9');
+insert into `t2` values (2, '2'), (4, '4'), (6, '6'), (8, '8'), (10, '10');
\ No newline at end of file