add index ddl not sync to downstream after restart downstream cluster

[aytrack]

What did you do?

1. deploy a bdr system with 1 primary cluster and 2 secondary clusters
2. create bdr changefeed from primary to secondary clusters
3. pause the changefeed
4. execute some add index ddl on primary cluster
5. resume the changefeed wait a few time restart 2 secondary cluster
6. check the ddl sync state

What did you expect to see?

All the ddls execute successfully on primary cluster should synced to secondary clusters

What did you see instead?

1. 1 ddl not synced to 1 of the secondary cluster
2. another secondary cluster executed all of the ddl

[2024/03/13 11:12:06.325 +00:00] [ERROR] [mysql_ddl_sink.go:224] ["Failed to rollback"] [sql="ALTER TABLE `test`.`xxx` ADD INDEX `L_H008_0014`(`a`, `b`, `c`)"] [namespace=default] [changefeed=cluster-a-cluster-b] [error="invalid connection"]
[2024/03/13 11:12:06.325 +00:00] [WARN] [mysql_ddl_sink.go:133] ["Execute DDL with error, retry later"] [startTs=448352302544453639] [ddl="ALTER TABLE `test`.`xxx` ADD INDEX `L_H008_0014`(`a`, `b`, `c`)"] [namespace=default] [changefeed=cluster-a-cluster-b] [error="invalid connection"]
[2024/03/13 11:12:06.835 +00:00] [INFO] [mysql_ddl_sink.go:192] ["Start exec DDL"] [DDL="ALTER TABLE `test`.`xxx` ADD INDEX `L_H008_0014`(`a`, `b`, `c`)"] [commitTs=448352302780383258] [namespace=default] [changefeed=cluster-a-cluster-b]
[2024/03/13 11:12:07.837 +00:00] [WARN] [mysql_ddl_sink.go:133] ["Execute DDL with error, retry later"] [startTs=448352302544453639] [ddl="ALTER TABLE `test`.`xxx` ADD INDEX `L_H008_0014`(`a`, `b`, `c`)"] [namespace=default] [changefeed=cluster-a-cluster-b] [error="dial tcp 10.103.134.244:4000: connect: connection refused"]
[2024/03/13 11:12:07.837 +00:00] [ERROR] [mysql_ddl_sink.go:291] ["async exec add index ddl failed"] [changefeedID=default/cluster-a-cluster-b] [commitTs=448352302780383258] [ddl="ALTER TABLE `test`.`xxx` ADD INDEX `L_H008_0014`(`a`, `b`, `c`)"]

Versions of the cluster

Upstream TiDB cluster version (execute SELECT tidb_version(); in a MySQL client):

[09:53:48]TiDB root:rtdb> select type,  version, git_hash from information_schema.cluster_info;
+-------+-------------+------------------------------------------+
| type  | version     | git_hash                                 |
+-------+-------------+------------------------------------------+
| tidb  | 8.0.0-alpha | 661f6d61aeeac5036c227dff7dcc9b179dec5381 |
| pd    | 8.0.0-alpha | c1eabda686e539e4bf13c166fda6b2ac29cced39 |
| tikv  | 8.0.0-alpha | 8ab73504d8abd5cea53b49e2f231c9fddcaafb66 |
| ticdc | 8.0.0-test  | d869a86fb086f12977f8c96993ea7ad9e2d1d69a |
+-------+-------------+------------------------------------------+