dispatcherManager: close dispatcher manager immediately even there is a long-time-cost ddl execution (#3634)

ref #3557

Author: hongyunyan

downstreamadapter/dispatchermanager/dispatcher_manager.go

@@ -861,17 +861,25 @@ func (e *DispatcherManager) close(removeChangefeed bool) {
 		e.heartBeatTask.Cancel()
 	}
 
+	// Cancel the context to signal all dependent components to stop.
+	// This is important to prevent `e.sink.Close() / e.sharedInfo.Close()` from blocking,
+	// especially when a long-running DDL is being executed by the sink.
+	e.cancel()
+
 	if e.sharedInfo != nil {
 		e.sharedInfo.Close()
 	}
 
+	log.Info("shared info closed", zap.Stringer("changefeedID", e.changefeedID))
+
 	if e.RedoEnable {
 		e.redoSink.Close(removeChangefeed)
 		// FIXME: cleanup redo log when remove the changefeed
 		e.closeRedoMeta(removeChangefeed)
 	}
 	e.sink.Close(removeChangefeed)
-	e.cancel()
+	log.Info("sink closed", zap.Stringer("changefeedID", e.changefeedID))
+
 	e.wg.Wait()
 
 	e.removeTaskHandles.Range(func(key, value interface{}) bool {

tests/integration_tests/pause_changefeed_with_long_time_ddl/run.sh

@@ -0,0 +1,51 @@
+#!/bin/bash
+# this test case is used to test whether we can successfully pause a changefeed when writing a long-time-cost ddl.
+
+set -eu
+
+CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+source $CUR/../_utils/test_prepare
+WORK_DIR=$OUT_DIR/$TEST_NAME
+CDC_BINARY=cdc.test
+SINK_TYPE=$1
+
+function run() {
+	# No need to test kafka and storage sink.
+	if [ "$SINK_TYPE" != "mysql" ]; then
+		return
+	fi
+	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
+
+	start_tidb_cluster --workdir $WORK_DIR
+
+	export GO_FAILPOINTS='github.com/pingcap/ticdc/pkg/sink/mysql/MySQLSinkExecDDLDelay=return("3600")'
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+
+	SINK_URI="mysql://root:@127.0.0.1:3306/"
+	cdc_cli_changefeed create --sink-uri="$SINK_URI" -c "test"
+
+	run_sql "use test;create table t1 (a int primary key, b int, unique key uk(b));" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+
+	cdc_cli_changefeed pause -c "test"
+
+	sleep 10
+
+	ensure 10 "check_logs_contains $WORK_DIR 'changefeed maintainer closed'"
+	check_table_not_exists test.t1 ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+
+	cdc_cli_changefeed resume -c "test"
+
+	cleanup_process $CDC_BINARY
+	export GO_FAILPOINTS=''
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+
+	check_table_exists test.t1 ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+
+	cleanup_process $CDC_BINARY
+	stop_tidb_cluster
+}
+
+trap 'stop_tidb_cluster; collect_logs $WORK_DIR' EXIT
+run $*
+check_logs $WORK_DIR
+echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>"

tests/integration_tests/run_light_it_in_ci.sh

@@ -60,7 +60,7 @@ mysql_groups=(
 	# G13
 	'cli_tls_with_auth cli_with_auth fail_over_ddl_N'
 	# G14
-	'batch_add_table batch_update_to_no_batch fail_over_ddl_O update_changefeed_check_config'
+	'batch_add_table batch_update_to_no_batch fail_over_ddl_O update_changefeed_check_config pause_changefeed_with_long_time_ddl'
 	# G15
 	'split_region changefeed_resume_with_checkpoint_ts autorandom gc_safepoint foreign_key_check old_arch_compatibility'
 )