dobule check before consumer reconnect (#1084)

apache · Oct 21, 2023 · da06dca · da06dca
1 parent 0bf2f9b
commit da06dca
Showing 1 changed file with 7 additions and 0 deletions.
diff --git 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