Repository: apache/pulsar-client-go
Pull request #1217 (2024-05-14 03:12:32)

[fix] Return an error when AckCumulative on a Shared/KeyShared subscription

<!--
### Contribution Checklist
  
  - Name the pull request in the form "[Issue XYZ][component] Title of the pull request", where *XYZ* should be replaced by the actual issue number.
    Skip *Issue XYZ* if there is no associated github issue for this pull request.
    Skip *component* if you are unsure about which is the best component. E.g. `[docs] Fix typo in produce method`.

  - Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
  
  - Each pull request should address only one issue, not mix up code from multiple issues.
  
  - Each commit in the pull request has a meaningful commit message

  - Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.

**(The sections below can be removed for hotfixes of typos)**
-->

### Motivation

The consumer should return error when AckCumulative on a Shared/KeyShared subscription


### Verifying this change

This change added tests

### Does this pull request potentially affect one of the following parts:

*If `yes` was chosen, please highlight the changes*

  - Dependencies (does it add or upgrade a dependency): (yes / no)
  - The public API: (yes / no)
  - The schema: (yes / no / don't know)
  - The default values of configurations: (yes / no)
  - The wire protocol: (yes / no)

### Documentation

  - Does this pull request introduce a new feature? (yes / no)
  - If yes, how is the feature documented? (not applicable / docs / GoDocs / not documented)
  - If a feature is not applicable for documentation, explain why?
  - If a feature is not documented yet in this PR, please create a followup issue for adding the documentation



Patch:
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index dc01e69..f752afb 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -20,7 +20,6 @@ package pulsar
 import (
 	"container/list"
 	"encoding/hex"
-	"errors"
 	"fmt"
 	"math"
 	"strings"
@@ -36,6 +35,7 @@ import (
 	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
 	"github.com/apache/pulsar-client-go/pulsar/log"
 	"github.com/bits-and-blooms/bitset"
+	"github.com/pkg/errors"
 
 	uAtomic "go.uber.org/atomic"
 )
@@ -50,6 +50,10 @@ const (
 	consumerClosed
 )
 
+var (
+	ErrInvalidAck = errors.New("invalid ack")
+)
+
 func (s consumerState) String() string {
 	switch s {
 	case consumerInit:
@@ -686,12 +690,20 @@ func (pc *partitionConsumer) AckIDWithResponseCumulative(msgID MessageID) error
 	return pc.internalAckIDCumulative(msgID, true)
 }
 
+func (pc *partitionConsumer) isAllowAckCumulative() bool {
+	return pc.options.subscriptionType != Shared && pc.options.subscriptionType != KeyShared
+}
+
 func (pc *partitionConsumer) internalAckIDCumulative(msgID MessageID, withResponse bool) error {
 	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
 		pc.log.WithField("state", state).Error("Failed to ack by closing or closed consumer")
 		return errors.New("consumer state is closed")
 	}
 
+	if !pc.isAllowAckCumulative() {
+		return errors.Wrap(ErrInvalidAck, "cumulative ack is not allowed for the Shared/KeyShared subscription type")
+	}
+
 	// chunk message id will be converted to tracking message id
 	trackingID := toTrackingMessageID(msgID)
 	if trackingID == nil {


Test patch:
diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
index 4120ba4..00f48ca 100644
--- a/pulsar/consumer_test.go
+++ b/pulsar/consumer_test.go
@@ -4392,3 +4392,38 @@ func TestMultiConsumerMemoryLimit(t *testing.T) {
 		return assert.Equal(t, pc2PrevQueueSize/2, pc2.currentQueueSize.Load())
 	})
 }
+
+func TestConsumerAckCumulativeOnSharedSubShouldFailed(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",
+		Type:             Shared,
+	})
+	assert.Nil(t, err)
+	defer consumer.Close()
+
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic: topic,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	_, err = producer.Send(context.Background(), &ProducerMessage{
+		Payload: []byte("hello"),
+	})
+	assert.Nil(t, err)
+
+	msg, err := consumer.Receive(context.Background())
+	assert.Nil(t, err)
+
+	err = consumer.AckIDCumulative(msg.ID())
+	assert.NotNil(t, err)
+	assert.ErrorIs(t, err, ErrInvalidAck)
+}