fix: SendAsync doesn't respect context and can't timeout during reconnection (#1422)

Author: RobertIndie

pulsar/internal/blocking_queue.go

@@ -24,8 +24,12 @@ import (
 // BlockingQueue is a interface of block queue
 type BlockingQueue interface {
 	// Put enqueue one item, block if the queue is full
+	// This is currently used for the internal testing
 	Put(item interface{})
 
+	// PutUnsafe enqueue one item without locking the queue, block if the queue is full
+	PutUnsafe(item interface{})
+
 	// Take dequeue one item, block until it's available
 	Take() interface{}
 
@@ -46,6 +50,17 @@ type BlockingQueue interface {
 
 	// ReadableSlice returns a new view of the readable items in the queue
 	ReadableSlice() []interface{}
+
+	// IterateUnsafe iterates the items in the queue without blocking the queue
+	IterateUnsafe(func(item interface{}))
+
+	// Lock locks the queue for manual control
+	// Users must call Unlock() after finishing their operations
+	Lock()
+
+	// Unlock unlocks the queue
+	// Must be called after Lock() to release the lock
+	Unlock()
 }
 
 type blockingQueue struct {
@@ -60,6 +75,17 @@ type blockingQueue struct {
 	isNotFull  *sync.Cond
 }
 
+func (bq *blockingQueue) IterateUnsafe(f func(item interface{})) {
+	readIdx := bq.headIdx
+	for i := 0; i < bq.size; i++ {
+		f(bq.items[readIdx])
+		readIdx++
+		if readIdx == bq.maxSize {
+			readIdx = 0
+		}
+	}
+}
+
 // NewBlockingQueue init block queue and returns a BlockingQueue
 func NewBlockingQueue(maxSize int) BlockingQueue {
 	bq := &blockingQueue{
@@ -76,9 +102,12 @@ func NewBlockingQueue(maxSize int) BlockingQueue {
 }
 
 func (bq *blockingQueue) Put(item interface{}) {
-	bq.mutex.Lock()
-	defer bq.mutex.Unlock()
+	bq.Lock()
+	defer bq.Unlock()
+	bq.PutUnsafe(item)
+}
 
+func (bq *blockingQueue) PutUnsafe(item interface{}) {
 	for bq.size == bq.maxSize {
 		bq.isNotFull.Wait()
 	}
@@ -192,3 +221,11 @@ func (bq *blockingQueue) ReadableSlice() []interface{} {
 
 	return res
 }
+
+func (bq *blockingQueue) Lock() {
+	bq.mutex.Lock()
+}
+
+func (bq *blockingQueue) Unlock() {
+	bq.mutex.Unlock()
+}

pulsar/internal/blocking_queue_test.go

@@ -148,3 +148,101 @@ func TestBlockingQueue_ReadableSlice(t *testing.T) {
 	assert.Equal(t, items[1], 3)
 	assert.Equal(t, items[2], 4)
 }
+
+func TestBlockingQueueIterate(t *testing.T) {
+	bq := NewBlockingQueue(5)
+
+	// Add some items
+	bq.PutUnsafe("item1")
+	bq.PutUnsafe("item2")
+	bq.PutUnsafe("item3")
+
+	// Test iteration
+	items := make([]interface{}, 0)
+	bq.IterateUnsafe(func(item interface{}) {
+		items = append(items, item)
+	})
+
+	assert.Equal(t, 3, len(items))
+	assert.Equal(t, "item1", items[0])
+	assert.Equal(t, "item2", items[1])
+	assert.Equal(t, "item3", items[2])
+}
+
+func TestBlockingQueueIteratePartial(t *testing.T) {
+	bq := NewBlockingQueue(5)
+
+	// Add some items
+	bq.PutUnsafe("item1")
+	bq.PutUnsafe("item2")
+	bq.PutUnsafe("item3")
+
+	// Test partial iteration (first 2 items only)
+	items := make([]interface{}, 0)
+	bq.IterateUnsafe(func(item interface{}) {
+		if len(items) < 2 {
+			items = append(items, item)
+		}
+	})
+
+	assert.Equal(t, 2, len(items))
+	assert.Equal(t, "item1", items[0])
+	assert.Equal(t, "item2", items[1])
+}
+
+func TestBlockingQueueIterateCircularBuffer(t *testing.T) {
+	bq := NewBlockingQueue(3)
+
+	// Fill the queue to test circular buffer behavior
+	bq.PutUnsafe("item1")
+	bq.PutUnsafe("item2")
+	bq.PutUnsafe("item3")
+
+	// Remove one item to create space
+	bq.Poll()
+
+	// Add another item to test wrapping
+	bq.PutUnsafe("item4")
+
+	// Test iteration with circular buffer
+	items := make([]interface{}, 0)
+	bq.IterateUnsafe(func(item interface{}) {
+		items = append(items, item)
+	})
+
+	assert.Equal(t, 3, len(items))
+	assert.Equal(t, "item2", items[0])
+	assert.Equal(t, "item3", items[1])
+	assert.Equal(t, "item4", items[2])
+}
+
+func TestBlockingQueueIterateEmpty(t *testing.T) {
+	bq := NewBlockingQueue(5)
+
+	// Test iteration on empty queue
+	items := make([]interface{}, 0)
+	bq.IterateUnsafe(func(item interface{}) {
+		items = append(items, item)
+	})
+
+	assert.Equal(t, 0, len(items))
+}
+
+func TestBlockingQueueManualLock(t *testing.T) {
+	bq := NewBlockingQueue(5)
+
+	// Test manual locking for batch PutUnsafe operations
+	bq.Lock()
+
+	bq.PutUnsafe("item1")
+	bq.PutUnsafe("item2")
+	bq.PutUnsafe("item3")
+
+	// Unlock
+	bq.Unlock()
+
+	// Verify all items were added
+	assert.Equal(t, 3, bq.Size())
+	assert.Equal(t, "item1", bq.Peek())
+	assert.Equal(t, "item3", bq.PeekLast())
+}

pulsar/producer.go

@@ -235,6 +235,15 @@ type Producer interface {
 	// This call is blocked when the `maxPendingMessages` becomes full (default: 1000)
 	// The callback will report back the message being published and
 	// the eventual error in publishing
+	// The context passed in the call is only used for the duration of the SendAsync call itself
+	// (i.e., to control blocking when the queue is full), and not for the entire message lifetime.
+	// Once SendAsync returns, the message lifetime is controlled by the SendTimeout configuration.
+	// Example:
+	// producer.SendAsync(ctx, &pulsar.ProducerMessage{
+	//     Payload: myPayload,
+	// }, func(msgID pulsar.MessageID, message *pulsar.ProducerMessage, err error) {
+	//     // handle publish result
+	// })
 	SendAsync(context.Context, *ProducerMessage, func(MessageID, *ProducerMessage, error))
 
 	// LastSequenceID get the last sequence id that was published by this producer.