[pulsar-client-go] branch master updated: Fix batched messages not ACKed correctly when batch index ACK is disabled (#994)

[ba...@apache.org]

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

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


The following commit(s) were added to refs/heads/master by this push:
     new 04ad521  Fix batched messages not ACKed correctly when batch index ACK is disabled (#994)
04ad521 is described below

commit 04ad521e6d061768ff02ac9bf024932d45b2bd94
Author: Yunze Xu <xy...@163.com>
AuthorDate: Mon Mar 20 22:18:16 2023 +0800

    Fix batched messages not ACKed correctly when batch index ACK is disabled (#994)
    
    * Fix batched messages not ACKed correctly when batch index ACK is disabled
    
    Fixes https://github.com/apache/pulsar-client-go/issues/993
    
    ### Motivation
    
    When batch index ACK is disabled, if N messages in a batch are
    acknowledged, currently only the batched message ID of the last message
    will be acknowledged. This behavior is wrong because we need to
    acknowledge the whole batch.
    
    ### Modifications
    
    - Create a `messageID` instance to ACK for this case
    - Add `TestConsumerBatchIndexAckDisabled` to cover this case
    
    * Fix ackRequest error
    
    * Fix wrong received msg id
    
    * Add comments
---
 pulsar/consumer_partition.go |  8 ++++++++
 pulsar/consumer_test.go      | 40 ++++++++++++++++++++++++++++++++++++++++
 2 files changed, 48 insertions(+)

diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index a08405a..087cec5 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -499,6 +499,14 @@ func (pc *partitionConsumer) ackID(msgID MessageID, withResponse bool) error {
 	trackingID := toTrackingMessageID(msgID)
 
 	if trackingID != nil && trackingID.ack() {
+		// All messages in the same batch have been acknowledged, we only need to acknowledge the
+		// MessageID that represents the entry that stores the whole batch
+		trackingID = &trackingMessageID{
+			messageID: &messageID{
+				ledgerID: trackingID.ledgerID,
+				entryID:  trackingID.entryID,
+			},
+		}
 		pc.metrics.AcksCounter.Inc()
 		pc.metrics.ProcessingTime.Observe(float64(time.Now().UnixNano()-trackingID.receivedTime.UnixNano()) / 1.0e9)
 	} else if !pc.options.enableBatchIndexAck {
diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
index a93dea9..1477ce6 100644
--- a/pulsar/consumer_test.go
+++ b/pulsar/consumer_test.go
@@ -4073,3 +4073,43 @@ func TestConsumerWithAutoScaledQueueReceive(t *testing.T) {
 		return assert.Equal(t, 3, int(pc.currentQueueSize.Load()))
 	})
 }
+
+func TestConsumerBatchIndexAckDisabled(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topic := newTopicName()
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:            topic,
+		SubscriptionName: "my-sub",
+	})
+	assert.Nil(t, err)
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic: topic,
+	})
+	assert.Nil(t, err)
+
+	for i := 0; i < 5; i++ {
+		producer.SendAsync(context.Background(), &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("msg-%d", i)),
+		}, nil)
+	}
+	for i := 0; i < 5; i++ {
+		message, err := consumer.Receive(context.Background())
+		assert.Nil(t, err)
+		consumer.Ack(message)
+	}
+	consumer.Close()
+	consumer, err = client.Subscribe(ConsumerOptions{
+		Topic:            topic,
+		SubscriptionName: "my-sub",
+	})
+	assert.Nil(t, err)
+	producer.Send(context.Background(), &ProducerMessage{Payload: []byte("done")})
+	message, err := consumer.Receive(context.Background())
+	assert.Nil(t, err)
+	assert.Equal(t, []byte("done"), message.Payload())
+}