Repository: apache/pulsar-client-go
Pull request #1216 (2024-05-13 14:53:57)

[cleanup] Remove AvroCodec from JSONSchema


### Modifications

`AvroCodec` is not used.
Remove `AvroCodec` from `JSONSchema`.

### Verifying this change

- [x] Make sure that the change passes the CI checks.

*(Please pick either of the following options)*

This change is a trivial rework / code cleanup without any test coverage.

*(or)*

This change is already covered by existing tests, such as *(please describe tests)*.

*(or)*

This change added tests and can be verified as follows:

*(example:)*
  - *Added integration tests for end-to-end deployment with large payloads (10MB)*
  - *Extended integration test for recovery after broker failure*

### 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): (no)
  - The public API: (no)
  - The schema: (no)
  - The default values of configurations: (no)
  - The wire protocol: (no)

### Documentation

  - Does this pull request introduce a new feature? (no)
  - If yes, how is the feature documented? (GoDocs)
  - 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 {
diff --git a/pulsar/schema.go b/pulsar/schema.go
index 3427fb2..ac2192d 100644
--- a/pulsar/schema.go
+++ b/pulsar/schema.go
@@ -142,7 +142,6 @@ func initAvroCodec(codec string) (*goavro.Codec, error) {
 }
 
 type JSONSchema struct {
-	AvroCodec
 	SchemaInfo
 }
 


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)
+}