[fix] Fix Infinite Loop in Reader's HasNext Function

pulsar/consumer_partition.go

@@ -570,15 +570,30 @@ func (pc *partitionConsumer) internalUnsubscribe(unsub *unsubscribeRequest) {
 
 func (pc *partitionConsumer) getLastMessageID() (*trackingMessageID, error) {
 	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
-		pc.log.WithField("state", state).Error("Failed to redeliver closing or closed consumer")
-		return nil, errors.New("failed to redeliver closing or closed consumer")
+		pc.log.WithField("state", state).Error("Failed to getLastMessageID for the closing or closed consumer")
+		return nil, errors.New("failed to getLastMessageID for the closing or closed consumer")
 	}
-	req := &getLastMsgIDRequest{doneCh: make(chan struct{})}
-	pc.eventsCh <- req
+	backoff := &internal.DefaultBackoff{}
+	request := func() (*trackingMessageID, error) {
+		req := &getLastMsgIDRequest{doneCh: make(chan struct{})}
+		pc.eventsCh <- req
 
-	// wait for the request to complete
-	<-req.doneCh
-	return req.msgID, req.err
+		// wait for the request to complete
+		<-req.doneCh
+		return req.msgID, req.err
+	}
+	for {
+		msgID, err := request()
+		if err == nil {
+			return msgID, nil
+		}
+		nextDelay := backoff.Next()
+		if backoff.IsMaxBackoffReached() {
+			return nil, fmt.Errorf("failed to getLastMessageID due to %w", err)
+		}
+		pc.log.WithError(err).Errorf("Failed to get last message id from broker, retrying in %v...", nextDelay)
+		time.Sleep(nextDelay)
+	}
 }
 
 func (pc *partitionConsumer) internalGetLastMessageID(req *getLastMsgIDRequest) {
@@ -1977,16 +1992,11 @@ func (pc *partitionConsumer) hasNext() bool {
 		return true
 	}
 
-	for {
-		lastMsgID, err := pc.getLastMessageID()
-		if err != nil {
-			pc.log.WithError(err).Error("Failed to get last message id from broker")
-			continue
-		} else {
-			pc.lastMessageInBroker = lastMsgID
-			break
-		}
+	lastMsgID, err := pc.getLastMessageID()
+	if err != nil {
+		return false
 	}
+	pc.lastMessageInBroker = lastMsgID
 
 	return pc.hasMoreMessages()
 }

pulsar/reader.go

@@ -113,6 +113,7 @@ type Reader interface {
 	Next(context.Context) (Message, error)
 
 	// HasNext checks if there is any message available to read from the current position
+	// If there is any errors, it will return false
 	HasNext() bool
 
 	// Close the reader and stop the broker to push more messages

pulsar/reader_test.go

@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/apache/pulsar-client-go/pulsar/crypto"
+	"github.com/apache/pulsar-client-go/pulsar/internal"
 	"github.com/apache/pulsar-client-go/pulsaradmin"
 	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/admin/config"
 	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
@@ -1023,3 +1024,58 @@ func createPartitionedTopic(topic string, n int) error {
 	}
 	return nil
 }
+
+func TestReaderHasNextFailed(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: serviceURL,
+	})
+	assert.Nil(t, err)
+	topic := newTopicName()
+	r, err := client.CreateReader(ReaderOptions{
+		Topic:          topic,
+		StartMessageID: EarliestMessageID(),
+	})
+	assert.Nil(t, err)
+	r.(*reader).c.consumers[0].state.Store(consumerClosing)
+	assert.False(t, r.HasNext())
+}
+
+func TestReaderHasNextRetryFailed(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: serviceURL,
+	})
+	assert.Nil(t, err)
+	topic := newTopicName()
+	r, err := client.CreateReader(ReaderOptions{
+		Topic:          topic,
+		StartMessageID: EarliestMessageID(),
+		// Retry the connection 10 seconds after it's disconnected.
+		BackoffPolicy: newTestBackoffPolicy(10*time.Second, 10*time.Second),
+	})
+	assert.Nil(t, err)
+
+	// Disconnect the connection
+	r.(*reader).c.consumers[0].conn.Load().(internal.Connection).Close()
+	minTimer := time.NewTimer(10 * time.Second) // Timer to check if r.HasNext() blocked for at least 10s
+	maxTimer := time.NewTimer(20 * time.Second) // Timer to ensure r.HasNext() doesn't block for more than 20s
+
+	done := make(chan bool)
+	go func() {
+		assert.False(t, r.HasNext())
+		done <- true
+	}()
+
+	select {
+	case <-maxTimer.C:
+		t.Fatal("r.HasNext() blocked for more than 20s")
+	case <-done:
+
+		if minTimer.Stop() {
+			t.Fatal("r.HasNext() did not block for at least 10s")
+		}
+		if !maxTimer.Stop() {
+			t.Fatal("r.HasNext() blocked for more than 20s")
+		}
+	}
+
+}