(pulsar-client-go) 01/06: fix: failTimeoutMessages cannot delete outdated messages (#1247)

[zi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

zike pushed a commit to branch branch-0.13.0
in repository https://gitbox.apache.org/repos/asf/pulsar-client-go.git

commit 13a78684ee4c6df78c852bc65539581a2caf1e36
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Mon Jul 22 18:32:07 2024 +0800

    fix: failTimeoutMessages cannot delete outdated messages (#1247)
    
    * fix: failTimeoutMessages cannot delete outdated messages
    
    * Fix slice pass
    
    (cherry picked from commit a42cc24f8ff4c00f4d2f901bf9c823b5b2c653e6)
---
 pulsar/producer_partition.go | 36 +++++++++++++++---------------------
 1 file changed, 15 insertions(+), 21 deletions(-)

diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index 1b2fae72..78f1f3cc 100755
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -372,8 +372,7 @@ func (p *partitionProducer) grabCnx(assignedBrokerURL string) error {
 				continue
 			}
 			pi := item.(*pendingItem)
-			// when resending pending batches, we update the sendAt timestamp and put to the back of queue
-			// to avoid pending item been removed by failTimeoutMessages and cause race condition
+			// when resending pending batches, we update the sendAt timestamp to record the metric.
 			pi.Lock()
 			pi.sentAt = time.Now()
 			pi.Unlock()
@@ -814,19 +813,14 @@ func (p *partitionProducer) internalSingleSend(
 		return
 	}
 
-	p.pendingQueue.Put(&pendingItem{
-		sentAt:       time.Now(),
-		buffer:       buffer,
-		sequenceID:   sid,
-		sendRequests: []interface{}{sr},
-	})
-	p._getConn().WriteData(buffer)
+	p.writeData(buffer, sid, []interface{}{sr})
 }
 
 type pendingItem struct {
 	sync.Mutex
 	buffer        internal.Buffer
 	sequenceID    uint64
+	createdAt     time.Time
 	sentAt        time.Time
 	sendRequests  []interface{}
 	isDone        bool
@@ -868,13 +862,19 @@ func (p *partitionProducer) internalFlushCurrentBatch() {
 		return
 	}
 
+	p.writeData(batchData, sequenceID, callbacks)
+}
+
+func (p *partitionProducer) writeData(buffer internal.Buffer, sequenceID uint64, callbacks []interface{}) {
+	now := time.Now()
 	p.pendingQueue.Put(&pendingItem{
-		sentAt:       time.Now(),
-		buffer:       batchData,
+		createdAt:    now,
+		sentAt:       now,
+		buffer:       buffer,
 		sequenceID:   sequenceID,
 		sendRequests: callbacks,
 	})
-	p._getConn().WriteData(batchData)
+	p._getConn().WriteData(buffer)
 }
 
 func (p *partitionProducer) failTimeoutMessages() {
@@ -898,7 +898,7 @@ func (p *partitionProducer) failTimeoutMessages() {
 			continue
 		}
 		oldestItem := item.(*pendingItem)
-		if nextWaiting := diff(oldestItem.sentAt); nextWaiting > 0 {
+		if nextWaiting := diff(oldestItem.createdAt); nextWaiting > 0 {
 			// none of these pending messages have timed out, wait and retry
 			t.Reset(nextWaiting)
 			continue
@@ -930,7 +930,7 @@ func (p *partitionProducer) failTimeoutMessages() {
 					pi := m.(*pendingItem)
 					pi.Lock()
 					defer pi.Unlock()
-					if nextWaiting := diff(pi.sentAt); nextWaiting > 0 {
+					if nextWaiting := diff(pi.createdAt); nextWaiting > 0 {
 						// current and subsequent items not timeout yet, stop iterating
 						tickerNeedWaiting = nextWaiting
 						return false
@@ -995,13 +995,7 @@ func (p *partitionProducer) internalFlushCurrentBatches() {
 		if b.BatchData == nil {
 			continue
 		}
-		p.pendingQueue.Put(&pendingItem{
-			sentAt:       time.Now(),
-			buffer:       b.BatchData,
-			sequenceID:   b.SequenceID,
-			sendRequests: b.Callbacks,
-		})
-		p._getConn().WriteData(b.BatchData)
+		p.writeData(b.BatchData, b.SequenceID, b.Callbacks)
 	}
 
 }