pulsar consumer stuck after broker restarted

[wk989898]

What did you do?

run Pulsar random broker failure test case

What did you expect to see?

No response

What did you see instead?

pulsar consumer stuck

[2026/04/15 07:42:22.460 +00:00] [INFO] [connection.go:257] ["Connecting to broker"] [remote_addr=pulsar://downstream-lb.cdc-testbed-tps-8116459-1-425:6650]
[2026/04/15 07:42:22.462 +00:00] [WARN] [connection.go:287] ["Failed to connect to broker."] [remote_addr=pulsar://downstream-lb.cdc-testbed-tps-8116459-1-425:6650] [error="dial tcp 10.105.103.57:6650: connect: connection refused"]
[2026/04/15 07:42:22.674 +00:00] [INFO] [connection.go:257] ["Connecting to broker"] [remote_addr=pulsar://downstream-lb.cdc-testbed-tps-8116459-1-425:6650]
[2026/04/15 07:42:22.675 +00:00] [WARN] [connection.go:287] ["Failed to connect to broker."] [remote_addr=pulsar://downstream-lb.cdc-testbed-tps-8116459-1-425:6650] [error="dial tcp 10.105.103.57:6650: connect: connection refused"]
[2026/04/15 07:42:23.145 +00:00] [INFO] [connection.go:257] ["Connecting to broker"] [remote_addr=pulsar://downstream-lb.cdc-testbed-tps-8116459-1-425:6650]
[2026/04/15 07:42:23.145 +00:00] [WARN] [connection.go:287] ["Failed to connect to broker."] [remote_addr=pulsar://downstream-lb.cdc-testbed-tps-8116459-1-425:6650] [error="dial tcp 10.105.103.57:6650: connect: connection refused"]
[2026/04/15 07:42:24.050 +00:00] [INFO] [connection.go:257] ["Connecting to broker"] [remote_addr=pulsar://downstream-lb.cdc-testbed-tps-8116459-1-425:6650]
[2026/04/15 07:42:24.051 +00:00] [INFO] [connection.go:295] ["TCP connection established"] [remote_addr=pulsar://downstream-lb.cdc-testbed-tps-8116459-1-425:6650] [local_addr=10.200.76.56:53886]
[2026/04/15 07:42:24.144 +00:00] [INFO] [connection.go:348] ["Connection is ready"] [remote_addr=pulsar://downstream-lb.cdc-testbed-tps-8116459-1-425:6650] [local_addr=10.200.76.56:53886]

Versions of the cluster

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

(paste TiDB cluster version here)

Upstream TiKV version (execute tikv-server --version):

(paste TiKV version here)

TiCDC version (execute cdc version):

(paste TiCDC version here)

[wk989898]

This is similar to apache/pulsar-client-go#698