Repository: apache/pulsar-client-go
Pull request #776 (2022-05-23 08:00:18)

Support ack response for Go SDK
Signed-off-by: xiaolongran <xiaolongran@tencent.com>

### Motivation

Follow https://github.com/apache/pulsar/pull/8161, support Ack Response Command for Go SDK

### Modifications

- Add `AckWithResponse` in `ConsumerOptions` struct




Patch:
diff --git a/pulsar/consumer.go b/pulsar/consumer.go
index dfe27c5..2df1637 100644
--- a/pulsar/consumer.go
+++ b/pulsar/consumer.go
@@ -182,6 +182,13 @@ type ConsumerOptions struct {
 	// > Notice: the NackBackoffPolicy will not work with `consumer.NackID(MessageID)`
 	// > because we are not able to get the redeliveryCount from the message ID.
 	NackBackoffPolicy NackBackoffPolicy
+
+	// AckWithResponse is a return value added to Ack Command, and its purpose is to confirm whether Ack Command
+	// is executed correctly on the Broker side. When set to true, the error information returned by the Ack
+	// method contains the return value of the Ack Command processed by the Broker side; when set to false, the
+	// error information of the Ack method only contains errors that may occur in the Go SDK's own processing.
+	// Default: false
+	AckWithResponse bool
 }
 
 // Consumer is an interface that abstracts behavior of Pulsar's consumer
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
index e36f040..e887538 100644
--- a/pulsar/consumer_impl.go
+++ b/pulsar/consumer_impl.go
@@ -361,6 +361,7 @@ func (c *consumer) internalTopicSubscribeToPartitions() error {
 				keySharedPolicy:            c.options.KeySharedPolicy,
 				schema:                     c.options.Schema,
 				decryption:                 c.options.Decryption,
+				ackWithResponse:            c.options.AckWithResponse,
 			}
 			cons, err := newPartitionConsumer(c, c.client, opts, c.messageCh, c.dlq, c.metrics)
 			ch <- ConsumerError{
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index 30ffcb2..06ccfd5 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -104,6 +104,7 @@ type partitionConsumerOpts struct {
 	keySharedPolicy            *KeySharedPolicy
 	schema                     Schema
 	decryption                 *MessageDecryptionInfo
+	ackWithResponse            bool
 }
 
 type partitionConsumer struct {
@@ -525,12 +526,22 @@ func (pc *partitionConsumer) internalAck(req *ackRequest) {
 		EntryId:  proto.Uint64(uint64(msgID.entryID)),
 	}
 
+	reqID := pc.client.rpcClient.NewRequestID()
 	cmdAck := &pb.CommandAck{
 		ConsumerId: proto.Uint64(pc.consumerID),
 		MessageId:  messageIDs,
 		AckType:    pb.CommandAck_Individual.Enum(),
 	}
 
+	if pc.options.ackWithResponse {
+		_, err := pc.client.rpcClient.RequestOnCnx(pc._getConn(), reqID, pb.BaseCommand_ACK, cmdAck)
+		if err != nil {
+			pc.log.WithError(err).Error("Ack with response error")
+			req.err = err
+		}
+		return
+	}
+
 	err := pc.client.rpcClient.RequestOnCnxNoWait(pc._getConn(), pb.BaseCommand_ACK, cmdAck)
 	if err != nil {
 		pc.log.Error("Connection was closed when request ack cmd")
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index da9a901..fa8d055 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -564,6 +564,9 @@ func (c *connection) internalReceivedCommand(cmd *pb.BaseCommand, headersAndPayl
 	case pb.BaseCommand_MESSAGE:
 		c.handleMessage(cmd.GetMessage(), headersAndPayload)
 
+	case pb.BaseCommand_ACK_RESPONSE:
+		c.handleAckResponse(cmd.GetAckResponse())
+
 	case pb.BaseCommand_PING:
 		c.handlePing()
 	case pb.BaseCommand_PONG:
@@ -676,6 +679,26 @@ func (c *connection) handleResponseError(serverError *pb.CommandError) {
 	request.callback(nil, errors.New(errMsg))
 }
 
+func (c *connection) handleAckResponse(ackResponse *pb.CommandAckResponse) {
+	requestID := ackResponse.GetRequestId()
+	consumerID := ackResponse.GetConsumerId()
+
+	request, ok := c.deletePendingRequest(requestID)
+	if !ok {
+		c.log.Warnf("AckResponse has complete when receive response! requestId : %d, consumerId : %d",
+			requestID, consumerID)
+		return
+	}
+
+	if ackResponse.GetMessage() == "" {
+		request.callback(nil, nil)
+		return
+	}
+
+	errMsg := fmt.Sprintf("ack response error: %s: %s", ackResponse.GetError(), ackResponse.GetMessage())
+	request.callback(nil, errors.New(errMsg))
+}
+
 func (c *connection) handleSendReceipt(response *pb.CommandSendReceipt) {
 	producerID := response.GetProducerId()
 
diff --git a/pulsar/internal/metrics.go b/pulsar/internal/metrics.go
index ec1a96e..1cab470 100644
--- a/pulsar/internal/metrics.go
+++ b/pulsar/internal/metrics.go
@@ -482,7 +482,10 @@ func NewMetricsProvider(metricsCardinality int, userDefinedLabels map[string]str
 
 func (mp *Metrics) GetLeveledMetrics(t string) *LeveledMetrics {
 	labels := make(map[string]string, 3)
-	tn, _ := ParseTopicName(t)
+	tn, err := ParseTopicName(t)
+	if err != nil {
+		return nil
+	}
 	topic := TopicNameWithoutPartitionPart(tn)
 	switch mp.metricsLevel {
 	case 4:
diff --git a/pulsar/internal/topic_name.go b/pulsar/internal/topic_name.go
index 86a1ebe..481e41f 100644
--- a/pulsar/internal/topic_name.go
+++ b/pulsar/internal/topic_name.go
@@ -107,6 +107,9 @@ func ParseTopicName(topic string) (*TopicName, error) {
 }
 
 func TopicNameWithoutPartitionPart(tn *TopicName) string {
+	if tn == nil {
+		return ""
+	}
 	if tn.Partition < 0 {
 		return tn.Name
 	}


Test patch:
diff --git a/pulsar/consumer_partition_test.go b/pulsar/consumer_partition_test.go
index e7fcd5d..2255e52 100644
--- a/pulsar/consumer_partition_test.go
+++ b/pulsar/consumer_partition_test.go
@@ -21,11 +21,9 @@ import (
 	"sync"
 	"testing"
 
+	"github.com/apache/pulsar-client-go/pulsar/internal"
 	"github.com/apache/pulsar-client-go/pulsar/internal/crypto"
-
 	"github.com/stretchr/testify/assert"
-
-	"github.com/apache/pulsar-client-go/pulsar/internal"
 )
 
 func TestSingleMessageIDNoAckTracker(t *testing.T) {
diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
index cadd8e4..0366884 100644
--- a/pulsar/consumer_test.go
+++ b/pulsar/consumer_test.go
@@ -1321,6 +1321,41 @@ func TestRLQ(t *testing.T) {
 	assert.Nil(t, checkMsg)
 }
 
+func TestAckWithResponse(t *testing.T) {
+	now := time.Now().Unix()
+	topic01 := fmt.Sprintf("persistent://public/default/topic-%d-01", now)
+	ctx := context.Background()
+
+	client, err := NewClient(ClientOptions{URL: lookupURL})
+	assert.Nil(t, err)
+	defer client.Close()
+
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:                       topic01,
+		SubscriptionName:            "my-sub",
+		Type:                        Shared,
+		SubscriptionInitialPosition: SubscriptionPositionEarliest,
+		AckWithResponse:             true,
+	})
+	assert.Nil(t, err)
+	defer consumer.Close()
+
+	producer01, err := client.CreateProducer(ProducerOptions{Topic: topic01})
+	assert.Nil(t, err)
+	defer producer01.Close()
+	for i := 0; i < 10; i++ {
+		_, err = producer01.Send(ctx, &ProducerMessage{Payload: []byte(fmt.Sprintf("MSG_01_%d", i))})
+		assert.Nil(t, err)
+	}
+
+	for i := 0; i < 10; i++ {
+		msg, err := consumer.Receive(ctx)
+		assert.Nil(t, err)
+		err = consumer.Ack(msg)
+		assert.Nil(t, err)
+	}
+}
+
 func TestRLQMultiTopics(t *testing.T) {
 	now := time.Now().Unix()
 	topic01 := fmt.Sprintf("persistent://public/default/topic-%d-1", now)
diff --git a/pulsar/internal/topic_name_test.go b/pulsar/internal/topic_name_test.go
index f08fcd0..ab6537b 100644
--- a/pulsar/internal/topic_name_test.go
+++ b/pulsar/internal/topic_name_test.go
@@ -104,20 +104,24 @@ func TestParseTopicNameErrors(t *testing.T) {
 
 func TestTopicNameWithoutPartitionPart(t *testing.T) {
 	tests := []struct {
-		tn       TopicName
+		tn       *TopicName
 		expected string
 	}{
 		{
-			tn:       TopicName{Name: "persistent://public/default/my-topic", Partition: -1},
+			tn:       &TopicName{Name: "persistent://public/default/my-topic", Partition: -1},
 			expected: "persistent://public/default/my-topic",
 		},
 		{
-			tn:       TopicName{Name: "persistent://public/default/my-topic-partition-0", Partition: 0},
+			tn:       &TopicName{Name: "persistent://public/default/my-topic-partition-0", Partition: 0},
 			expected: "persistent://public/default/my-topic",
 		},
+		{
+			tn:       nil,
+			expected: "",
+		},
 	}
 	for _, test := range tests {
-		assert.Equal(t, test.expected, TopicNameWithoutPartitionPart(&test.tn))
+		assert.Equal(t, test.expected, TopicNameWithoutPartitionPart(test.tn))
 	}
 }