fix: track reconnect count in zero queue consumer

Author: crossoverJie

pulsar/consumer_partition.go

@@ -171,6 +171,7 @@ type partitionConsumer struct {
 	currentQueueSize       uAtomic.Int32
 	scaleReceiverQueueHint uAtomic.Bool
 	incomingMessages       uAtomic.Int32
+	reconnectCount         uAtomic.Int32
 
 	eventsCh        chan interface{}
 	connectedCh     chan struct{}
@@ -1544,6 +1545,7 @@ func createEncryptionContext(msgMeta *pb.MessageMetadata) *EncryptionContext {
 func (pc *partitionConsumer) ConnectionClosed(closeConsumer *pb.CommandCloseConsumer) {
 	// Trigger reconnection in the consumer goroutine
 	pc.log.Debug("connection closed and send to connectClosedCh")
+	pc.reconnectCount.Inc()
 	var assignedBrokerURL string
 	if closeConsumer != nil {
 		assignedBrokerURL = pc.client.selectServiceURL(

pulsar/consumer_zero_queue_test.go

@@ -318,7 +318,13 @@ func TestReconnectedBrokerSendPermits(t *testing.T) {
 	err = admin.Topics().Unload(*topicName)
 	assert.Nil(t, err)
 	log.Println("unloaded topic")
-	time.Sleep(1 * time.Minute)
+	zc, ok := consumer.(*zeroQueueConsumer)
+	assert.True(t, ok)
+	require.EventuallyWithT(t, func(c *assert.CollectT) {
+		reconnectCount := zc.pc.reconnectCount.Load()
+		require.Equal(c, reconnectCount, int32(1))
+	}, 30*time.Second, 1*time.Second)
+	//time.Sleep(1 * time.Minute)
 
 	// receive 10 messages
 	for i := 0; i < 10; i++ {