From 2626d2325206fd32149c256111b76e6f2ed6e0b4 Mon Sep 17 00:00:00 2001
From: Scepter <zc_cold@hotmail.com>
Date: Thu, 24 Aug 2023 10:58:00 +0800
Subject: [PATCH] dobule check before consumer reconnect

---
 pulsar/consumer_partition.go | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index 364dae754c..2ba842744c 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -1644,6 +1644,13 @@ func (pc *partitionConsumer) reconnectToBroker() {
 		pc.log.Info("Reconnecting to broker in ", delayReconnectTime)
 		time.Sleep(delayReconnectTime)
 
+		// double check
+		if pc.getConsumerState() != consumerReady {
+			// Consumer is already closing
+			pc.log.Info("consumer state not ready, exit reconnect")
+			return
+		}
+
 		err := pc.grabConn()
 		if err == nil {
 			// Successfully reconnected