kafka dml worker batch failed

[wk989898]

Reproduce

make integration_test_kafka CASE=kafka_big_messages

Log

[2025/01/14 15:44:45.389 +08:00] [INFO] [kafka_dml_worker.go:221] ["MQ sink batch worker started"] [namespace=default] [changefeed=sink-uri-verify-changefeed-id] [protocol=open-protocol]
[2025/01/14 15:44:45.390 +08:00] [INFO] [factory.go:222] ["async producer exit since context is done"] [namespace=default] [changefeed=sink-uri-verify-changefeed-id]
[2025/01/14 15:44:45.390 +08:00] [INFO] [factory.go:151] ["Kafka DDL producer closed"] [namespace=default] [changefeed=sink-uri-verify-changefeed-id] [duration=336.236µs]
[2025/01/14 15:44:45.390 +08:00] [DEBUG] [sarama.go:129] ["Closed connection to broker 127.0.0.1:9092"] [component=sarama]
[2025/01/14 15:44:45.390 +08:00] [ERROR] [kafka_dml_worker.go:240] ["kafka dml worker batch failed"] [namespace=default] [changefeed=sink-uri-verify-changefeed-id] [error="context canceled"]
[2025/01/14 15:44:45.390 +08:00] [DEBUG] [sarama.go:129] ["Closed connection to broker zhangxiaoshuai-wangkun.lxd:9092"] [component=sarama]
[2025/01/14 15:44:45.390 +08:00] [ERROR] [kafka_dml_worker.go:240] ["kafka dml worker batch failed"] [namespace=default] [changefeed=sink-uri-verify-changefeed-id] [error="context canceled"]
[2025/01/14 15:44:45.390 +08:00] [ERROR] [kafka_dml_worker.go:240] ["kafka dml worker batch failed"] [namespace=default] [changefeed=sink-uri-verify-changefeed-id] [error="context canceled"]
[2025/01/14 15:44:45.390 +08:00] [ERROR] [kafka_dml_worker.go:240] ["kafka dml worker batch failed"] [namespace=default] [changefeed=sink-uri-verify-changefeed-id] [error="context canceled"]
[2025/01/14 15:44:45.390 +08:00] [ERROR] [kafka_dml_worker.go:240] ["kafka dml worker batch failed"] [namespace=default] [changefeed=sink-uri-verify-changefeed-id] [error="context canceled"]
[2025/01/14 15:44:45.390 +08:00] [ERROR] [kafka_dml_worker.go:240] ["kafka dml worker batch failed"] [namespace=default] [changefeed=sink-uri-verify-changefeed-id] [error="context canceled"]
[2025/01/14 15:44:45.390 +08:00] [DEBUG] [sarama.go:132] ["Closing Client"] [component=sarama]
[2025/01/14 15:44:45.390 +08:00] [INFO] [kafka_topic_manager.go:121] ["Background refresh Kafka metadata goroutine exit."] [namespace=default] [changefeed=sink-uri-verify-changefeed-id]