Repository: apache/pulsar-client-go

Issue:
[Bug][Producer] The callback was not invoked during reconnecting.
#### Expected behavior

Callback should be called with a timeout error.

#### Actual behavior

Callback was not called

#### Steps to reproduce

1. Connect to the server;
2. Call SendAsync() to send a message, it will send successfully;
3. Stop the server;
4. Call SendAsync() to send some messages, the messages' callback won't be called anymore

#### System configuration
**Pulsar version**: x.y


Discussion:
@gunli: > I don't know if there is a same bug with the consumer.

It seems that when we call `SendAsync()` at this time, the messages are sent into `p.dataChan`, but `runEventsLoop()` is busy in doing reconnecting: `p.reconnectToBroker(connectionClosed)`, and have no chance to chandle the case `case data, ok := <-p.dataChan`, and `failTimeoutMessages()` just check the messages in `p.pendingQueue`, so no one can tell the messages are timeout.

```go
func (p *partitionProducer) runEventsLoop() {
	for {
		select {
		case data, ok := <-p.dataChan:
			// when doClose() is call, p.dataChan will be closed, data will be nil
			if !ok {
				return
			}
			p.internalSend(data)
		case cmd, ok := <-p.cmdChan:
			// when doClose() is call, p.dataChan will be closed, cmd will be nil
			if !ok {
				return
			}
			switch v := cmd.(type) {
			case *flushRequest:
				p.internalFlush(v)
			case *closeProducer:
				p.internalClose(v)
				return
			}
		case connectionClosed := <-p.connectClosedCh:
			p.log.Info("runEventsLoop will reconnect in producer")
			p.reconnectToBroker(connectionClosed)
		case <-p.batchFlushTicker.C:
			p.internalFlushCurrentBatch()
		}
	}
}

func (p *partitionProducer) failTimeoutMessages() {
	diff := func(sentAt time.Time) time.Duration {
		return p.options.SendTimeout - time.Since(sentAt)
	}

	t := time.NewTimer(p.options.SendTimeout)
	defer t.Stop()

	for range t.C {
		state := p.getProducerState()
		if state == producerClosing || state == producerClosed {
			return
		}

		item := p.pendingQueue.Peek()
		if item == nil {
			// pending queue is empty
			t.Reset(p.options.SendTimeout)
			continue
		}
		oldestItem := item.(*pendingItem)
		if nextWaiting := diff(oldestItem.createdAt); nextWaiting > 0 {
			// none of these pending messages have timed out, wait and retry
			t.Reset(nextWaiting)
			continue
		}

		// since pending queue is not thread safe because of there is no global iteration lock
		// to control poll from pending queue, current goroutine and connection receipt handler
		// iterate pending queue at the same time, this maybe a performance trade-off
		// see https://github.com/apache/pulsar-client-go/pull/301
		curViewItems := p.pendingQueue.ReadableSlice()
		viewSize := len(curViewItems)
		if viewSize <= 0 {
			// double check
			t.Reset(p.options.SendTimeout)
			continue
		}
		p.log.Infof("Failing %d messages on timeout %s", viewSize, p.options.SendTimeout)
		lastViewItem := curViewItems[viewSize-1].(*pendingItem)

		// iterate at most viewSize items
		for i := 0; i < viewSize; i++ {
			tickerNeedWaiting := time.Duration(0)
			item := p.pendingQueue.CompareAndPoll(
				func(m interface{}) bool {
					if m == nil {
						return false
					}

					pi := m.(*pendingItem)
					pi.Lock()
					defer pi.Unlock()
					if nextWaiting := diff(pi.createdAt); nextWaiting > 0 {
						// current and subsequent items not timeout yet, stop iterating
						tickerNeedWaiting = nextWaiting
						return false
					}
					return true
				})

			if item == nil {
				t.Reset(p.options.SendTimeout)
				break
			}

			if tickerNeedWaiting > 0 {
				t.Reset(tickerNeedWaiting)
				break
			}

			pi := item.(*pendingItem)
			pi.Lock()

			for _, i := range pi.sendRequests {
				sr := i.(*sendRequest)
				sr.done(nil, ErrSendTimeout)
			}

			// flag the sending has completed with error, flush make no effect
			pi.done(ErrSendTimeout)
			pi.Unlock()

			// finally reached the last view item, current iteration ends
			if pi == lastViewItem {
				t.Reset(p.options.SendTimeout)
				break
			}
		}
	}
}
```@gunli: @shibd @nodece @RobertIndie @BewareMyPower @crossoverJie @gunli: We have discussed whether we should block the send and flush requests when we are reconnecting in #1249, but not sufficient enough.
Now, IMO, we should not block send/flush when we are reconnecting, 'cause it has no side effect,  send/flush will succeed or timeout, if we block them, as this bug described, timeout will not triggered.@BewareMyPower: What's your client and producer configs? There was a similar issue happened when `maxReconnectAttempts` is 1: https://github.com/apache/pulsar-client-go/issues/1312#issuecomment-2597791188 @gunli: > What's your client and producer configs? There was a similar issue happened when `maxReconnectAttempts` is 1: [#1312 (comment)](https://github.com/apache/pulsar-client-go/issues/1312#issuecomment-2597791188)

My maxReconnectAttempts  is the default value: -1

Solution patch:
diff --git a/pulsar/internal/blocking_queue.go b/pulsar/internal/blocking_queue.go
index b44ec16..a4b9cd9 100644
--- a/pulsar/internal/blocking_queue.go
+++ b/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()
+}
diff --git a/pulsar/producer.go b/pulsar/producer.go
index 76ab853..633e761 100644
--- a/pulsar/producer.go
+++ b/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.
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index 9841222..a6b83a8 100755
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -47,6 +47,7 @@ type producerState int32
 const (
 	// producer states
 	producerInit = iota
+	producerConnecting
 	producerReady
 	producerClosing
 	producerClosed
@@ -218,6 +219,7 @@ func newPartitionProducer(client *client, topic string, options *ProducerOptions
 	} else {
 		p.userProvidedProducerName = false
 	}
+	p.setProducerState(producerConnecting)
 	err := p.grabCnx("")
 	if err != nil {
 		p.batchFlushTicker.Stop()
@@ -265,6 +267,10 @@ func (p *partitionProducer) lookupTopic(brokerServiceURL string) (*internal.Look
 }
 
 func (p *partitionProducer) grabCnx(assignedBrokerURL string) error {
+	if !p.casProducerState(producerReady, producerConnecting) && p.isClosingOrClosed() {
+		// closing or closed
+		return ErrProducerClosed
+	}
 	lr, err := p.lookupTopic(assignedBrokerURL)
 	if err != nil {
 		return err
@@ -385,31 +391,20 @@ func (p *partitionProducer) grabCnx(assignedBrokerURL string) error {
 		"epoch": atomic.LoadUint64(&p.epoch),
 	}).Info("Connected producer")
 
-	pendingItems := p.pendingQueue.ReadableSlice()
-	viewSize := len(pendingItems)
-	if viewSize > 0 {
-		p.log.Infof("Resending %d pending batches", viewSize)
-		lastViewItem := pendingItems[viewSize-1].(*pendingItem)
-
-		// iterate at most pending items
-		for i := 0; i < viewSize; i++ {
-			item := p.pendingQueue.Poll()
-			if item == nil {
-				continue
-			}
-			pi := item.(*pendingItem)
-			// when resending pending batches, we update the sendAt timestamp to record the metric.
-			pi.Lock()
-			pi.sentAt = time.Now()
-			pi.Unlock()
-			pi.buffer.Retain() // Retain for writing to the connection
-			p.pendingQueue.Put(pi)
-			p._getConn().WriteData(pi.ctx, pi.buffer)
+	p.pendingQueue.Lock()
+	defer p.pendingQueue.Unlock()
+	p.pendingQueue.IterateUnsafe(func(item any) {
+		pi := item.(*pendingItem)
+		// when resending pending batches, we update the sendAt timestamp to record the metric.
+		pi.Lock()
+		pi.sentAt = time.Now()
+		pi.Unlock()
+		pi.buffer.Retain() // Retain for writing to the connection
+		p._getConn().WriteData(pi.ctx, pi.buffer)
+	})
 
-			if pi == lastViewItem {
-				break
-			}
-		}
+	if !p.casProducerState(producerConnecting, producerReady) && p.isClosingOrClosed() {
+		return ErrProducerClosed
 	}
 	return nil
 }
@@ -495,9 +490,9 @@ func (p *partitionProducer) reconnectToBroker(connectionClosed *connectionClosed
 			return struct{}{}, nil
 		}
 
-		if p.getProducerState() != producerReady {
+		if p.isClosingOrClosed() {
 			// Producer is already closing
-			p.log.Info("producer state not ready, exit reconnect")
+			p.log.Info("producer state is in closing or closed, exit reconnect")
 			return struct{}{}, nil
 		}
 
@@ -561,6 +556,18 @@ func (p *partitionProducer) reconnectToBroker(connectionClosed *connectionClosed
 }
 
 func (p *partitionProducer) runEventsLoop() {
+	go func() {
+		for {
+			select {
+			case connectionClosed := <-p.connectClosedCh:
+				p.log.Info("runEventsLoop will reconnect in producer")
+				p.reconnectToBroker(connectionClosed)
+			case <-p.ctx.Done():
+				p.log.Info("Producer is shutting down. Close the reconnect event loop")
+				return
+			}
+		}
+	}()
 	for {
 		select {
 		case data, ok := <-p.dataChan:
@@ -581,9 +588,6 @@ func (p *partitionProducer) runEventsLoop() {
 				p.internalClose(v)
 				return
 			}
-		case connectionClosed := <-p.connectClosedCh:
-			p.log.Info("runEventsLoop will reconnect in producer")
-			p.reconnectToBroker(connectionClosed)
 		case <-p.batchFlushTicker.C:
 			p.internalFlushCurrentBatch()
 		}
@@ -902,7 +906,17 @@ func (p *partitionProducer) writeData(buffer internal.Buffer, sequenceID uint64,
 		now := time.Now()
 		ctx, cancel := context.WithCancel(context.Background())
 		buffer.Retain()
-		p.pendingQueue.Put(&pendingItem{
+		p.pendingQueue.Lock()
+		defer p.pendingQueue.Unlock()
+		conn := p._getConn()
+		if p.getProducerState() == producerReady {
+			// If the producer is reconnecting, we should not write to the connection.
+			// We just need to push the buffer to the pending queue, it will be sent during the reconnecting.
+			conn.WriteData(ctx, buffer)
+		} else {
+			p.log.Debug("Skipping write to connection, producer state: ", p.getProducerState())
+		}
+		p.pendingQueue.PutUnsafe(&pendingItem{
 			ctx:          ctx,
 			cancel:       cancel,
 			createdAt:    now,
@@ -911,7 +925,6 @@ func (p *partitionProducer) writeData(buffer internal.Buffer, sequenceID uint64,
 			sequenceID:   sequenceID,
 			sendRequests: callbacks,
 		})
-		p._getConn().WriteData(ctx, buffer)
 	}
 }
 
@@ -924,8 +937,7 @@ func (p *partitionProducer) failTimeoutMessages() {
 	defer t.Stop()
 
 	for range t.C {
-		state := p.getProducerState()
-		if state == producerClosing || state == producerClosed {
+		if p.isClosingOrClosed() {
 			return
 		}
 
@@ -1323,11 +1335,16 @@ func (p *partitionProducer) internalSendAsync(
 		return
 	}
 
-	if p.getProducerState() != producerReady {
+	if p.isClosingOrClosed() {
 		sr.done(nil, ErrProducerClosed)
 		return
 	}
 
+	if err := ctx.Err(); err != nil {
+		sr.done(nil, ctx.Err())
+		return
+	}
+
 	p.options.Interceptors.BeforeSend(p, msg)
 
 	if err := p.updateSchema(sr); err != nil {
@@ -1356,7 +1373,13 @@ func (p *partitionProducer) internalSendAsync(
 		return
 	}
 
-	p.dataChan <- sr
+	select {
+	case <-ctx.Done():
+		sr.done(nil, ctx.Err())
+		return
+	case p.dataChan <- sr:
+		return
+	}
 }
 
 func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt) {
@@ -1441,7 +1464,7 @@ func (p *partitionProducer) internalClose(req *closeProducer) {
 }
 
 func (p *partitionProducer) doClose(reason error) {
-	if !p.casProducerState(producerReady, producerClosing) {
+	if !p.casProducerState(producerReady, producerClosing) && !p.casProducerState(producerConnecting, producerClosing) {
 		return
 	}
 
@@ -1522,7 +1545,7 @@ func (p *partitionProducer) Flush() error {
 }
 
 func (p *partitionProducer) FlushWithCtx(ctx context.Context) error {
-	if p.getProducerState() != producerReady {
+	if p.isClosingOrClosed() {
 		return ErrProducerClosed
 	}
 
@@ -1549,6 +1572,11 @@ func (p *partitionProducer) getProducerState() producerState {
 	return producerState(p.state.Load())
 }
 
+func (p *partitionProducer) isClosingOrClosed() bool {
+	state := p.getProducerState()
+	return state == producerClosing || state == producerClosed
+}
+
 func (p *partitionProducer) setProducerState(state producerState) {
 	p.state.Swap(int32(state))
 }
@@ -1560,8 +1588,8 @@ func (p *partitionProducer) casProducerState(oldState, newState producerState) b
 }
 
 func (p *partitionProducer) Close() {
-	if p.getProducerState() != producerReady {
-		// Producer is closing
+	if p.isClosingOrClosed() {
+		// Producer is closing or closed
 		return
 	}
 



Test patch:
diff --git a/pulsar/internal/blocking_queue_test.go b/pulsar/internal/blocking_queue_test.go
index c93b1a6..8acd282 100644
--- a/pulsar/internal/blocking_queue_test.go
+++ b/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())
+}
diff --git a/pulsar/producer_test.go b/pulsar/producer_test.go
index 02b4b67..0087607 100644
--- a/pulsar/producer_test.go
+++ b/pulsar/producer_test.go
@@ -58,7 +58,8 @@ func TestInvalidURL(t *testing.T) {
 
 func TestProducerConnectError(t *testing.T) {
 	client, err := NewClient(ClientOptions{
-		URL: "pulsar://invalid-hostname:6650",
+		URL:              "pulsar://invalid-hostname:6650",
+		OperationTimeout: 3 * time.Second,
 	})
 
 	assert.Nil(t, err)
@@ -348,7 +349,7 @@ func TestFlushInProducer(t *testing.T) {
 	assert.NoError(t, err)
 	defer client.Close()
 
-	topicName := "test-flush-in-producer"
+	topicName := newTopicName()
 	subName := "subscription-name"
 	numOfMessages := 10
 	ctx := context.Background()
@@ -2334,7 +2335,7 @@ func TestMemLimitContextCancel(t *testing.T) {
 			Payload: make([]byte, 1024),
 		}, func(_ MessageID, _ *ProducerMessage, e error) {
 			assert.Error(t, e)
-			assert.ErrorContains(t, e, getResultStr(TimeoutError))
+			assert.ErrorContains(t, e, "context canceled")
 			wg.Done()
 		})
 	}()
@@ -2353,6 +2354,32 @@ func TestMemLimitContextCancel(t *testing.T) {
 	assert.NoError(t, err)
 }
 
+func TestSendAsyncWithContextCancel(t *testing.T) {
+
+	c, err := NewClient(ClientOptions{
+		URL:              serviceURL,
+		MemoryLimitBytes: 100 * 1024,
+	})
+	assert.NoError(t, err)
+	defer c.Close()
+
+	topicName := newTopicName()
+	producer, _ := c.CreateProducer(ProducerOptions{
+		Topic: topicName,
+	})
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	var callbackErr error
+	producer.SendAsync(ctx, &ProducerMessage{
+		Payload: make([]byte, 1024),
+	}, func(_ MessageID, _ *ProducerMessage, e error) {
+		callbackErr = e
+	})
+
+	require.Error(t, callbackErr)
+}
+
 func TestBatchSendMessagesWithMetadata(t *testing.T) {
 	testSendMessagesWithMetadata(t, false)
 }
@@ -2500,6 +2527,15 @@ func (pqw *pendingQueueWrapper) Put(item interface{}) {
 	pqw.pendingQueue.Put(item)
 }
 
+func (pqw *pendingQueueWrapper) PutUnsafe(item interface{}) {
+	pi := item.(*pendingItem)
+	writerIdx := pi.buffer.WriterIndex()
+	buf := internal.NewBuffer(int(writerIdx))
+	buf.Write(pi.buffer.Get(0, writerIdx))
+	*pqw.writtenBuffers = append(*pqw.writtenBuffers, buf)
+	pqw.pendingQueue.PutUnsafe(item)
+}
+
 func (pqw *pendingQueueWrapper) Take() interface{} {
 	return pqw.pendingQueue.Take()
 }
@@ -2528,6 +2564,18 @@ func (pqw *pendingQueueWrapper) ReadableSlice() []interface{} {
 	return pqw.pendingQueue.ReadableSlice()
 }
 
+func (pqw *pendingQueueWrapper) IterateUnsafe(f func(item interface{})) {
+	pqw.pendingQueue.IterateUnsafe(f)
+}
+
+func (pqw *pendingQueueWrapper) Lock() {
+	pqw.pendingQueue.Lock()
+}
+
+func (pqw *pendingQueueWrapper) Unlock() {
+	pqw.pendingQueue.Unlock()
+}
+
 func TestDisableReplication(t *testing.T) {
 	client, err := NewClient(ClientOptions{
 		URL: serviceURL,
@@ -2741,3 +2789,112 @@ func TestSendBufferRetainWhenConnectionStuck(t *testing.T) {
 	b := conn.buffers[0]
 	assert.Equal(t, int64(1), b.RefCnt(), "Expected buffer to have a reference count of 1 after sending")
 }
+
+func TestSendAsyncCouldTimeoutWhileReconnecting(t *testing.T) {
+	testSendAsyncCouldTimeoutWhileReconnecting(t, false)
+	testSendAsyncCouldTimeoutWhileReconnecting(t, true)
+}
+
+func testSendAsyncCouldTimeoutWhileReconnecting(t *testing.T, isDisableBatching bool) {
+	t.Helper()
+
+	req := testcontainers.ContainerRequest{
+		Image:        getPulsarTestImage(),
+		ExposedPorts: []string{"6650/tcp", "8080/tcp"},
+		WaitingFor:   wait.ForExposedPort(),
+		Cmd:          []string{"bin/pulsar", "standalone", "-nfw"},
+	}
+	c, err := testcontainers.GenericContainer(context.Background(), testcontainers.GenericContainerRequest{
+		ContainerRequest: req,
+		Started:          true,
+	})
+	require.NoError(t, err, "Failed to start the pulsar container")
+	defer func() {
+		err := c.Terminate(context.Background())
+		if err != nil {
+			t.Fatal("Failed to terminate the pulsar container", err)
+		}
+	}()
+
+	endpoint, err := c.PortEndpoint(context.Background(), "6650", "pulsar")
+	require.NoError(t, err, "Failed to get the pulsar endpoint")
+
+	client, err := NewClient(ClientOptions{
+		URL:               endpoint,
+		ConnectionTimeout: 5 * time.Second,
+		OperationTimeout:  5 * time.Second,
+	})
+	require.NoError(t, err)
+	defer client.Close()
+
+	var testProducer Producer
+	require.Eventually(t, func() bool {
+		testProducer, err = client.CreateProducer(ProducerOptions{
+			Topic:               newTopicName(),
+			Schema:              NewBytesSchema(nil),
+			SendTimeout:         3 * time.Second,
+			DisableBatching:     isDisableBatching,
+			BatchingMaxMessages: 5,
+			MaxPendingMessages:  10,
+		})
+		return err == nil
+	}, 30*time.Second, 1*time.Second)
+
+	numMessages := 10
+	// Send 10 messages synchronously
+	for i := 0; i < numMessages; i++ {
+		send, err := testProducer.Send(context.Background(), &ProducerMessage{Payload: []byte("test")})
+		require.NoError(t, err)
+		require.NotNil(t, send)
+	}
+
+	// stop pulsar server
+	timeout := 10 * time.Second
+	err = c.Stop(context.Background(), &timeout)
+	require.NoError(t, err)
+
+	// Test the SendAsync could be timeout if the producer is reconnecting
+
+	finalErr := make(chan error, 1)
+	testProducer.SendAsync(context.Background(), &ProducerMessage{
+		Payload: []byte("test"),
+	}, func(_ MessageID, _ *ProducerMessage, err error) {
+		finalErr <- err
+	})
+	select {
+	case <-time.After(10 * time.Second):
+		t.Fatal("test timeout")
+	case err = <-finalErr:
+		// should get a timeout error
+		require.ErrorIs(t, err, ErrSendTimeout)
+	}
+	close(finalErr)
+
+	// Test that the SendAsync could be timeout if the pending queue is full
+
+	go func() {
+		// Send 10 messages asynchronously to make the pending queue full
+		for i := 0; i < numMessages; i++ {
+			testProducer.SendAsync(context.Background(), &ProducerMessage{
+				Payload: []byte("test"),
+			}, func(_ MessageID, _ *ProducerMessage, _ error) {
+			})
+		}
+	}()
+
+	time.Sleep(3 * time.Second)
+	finalErr = make(chan error, 1)
+	testProducer.SendAsync(context.Background(), &ProducerMessage{
+		Payload: []byte("test"),
+	}, func(_ MessageID, _ *ProducerMessage, err error) {
+		finalErr <- err
+	})
+	select {
+	case <-time.After(10 * time.Second):
+		t.Fatal("test timeout")
+	case err = <-finalErr:
+		// should get a timeout error
+		require.ErrorIs(t, err, ErrSendTimeout)
+	}
+	close(finalErr)
+}