ddl_manager (ticdc): fix in bdr mode cdc can not replicate a table's dmls after drop and re-create it (#10080)

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

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));

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');