Repository: apache/pulsar-client-go (go)

Issue:
Allow CustomProperties when calling ReconsumeLater
**Is your feature request related to a problem? Please describe.**
Go SDK is missing custome properties when calling `ReconsumeLater()` like java sdk

**Describe the solution you'd like**
We should allow to add cusomte properies when a `ReconsumeLater` function is being called

**Describe alternatives you've considered**
We are not able to identify as of now. We are creating new message and adding custome properies at run time.



Pull request description:
Allow CustomProperties when sending message for retry
<!--
### 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)**
-->

*(If this PR fixes a github issue, please add `Fixes #<xyz>`.)*

Fixes #915 

*(or if this PR is one task of a github issue, please add `Master Issue: #<xyz>` to link to the master issue.)*

Master Issue: #915 

### Motivation
In Java SDK custom properties are allowed when ReconsumeLater is called so adding same in golang

### Modifications
Created a function which will take custom message properties and add them to retry message as per the doc.
To reduce the same code i callback this function from the reconsume later function. because other part of the code is same execpt the one which should change.

### 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:  may be, added another public function for consumer to allow custom properties
  - The schema: (yes / no / don't know)
  - The default values of configurations: no
  - The wire protocol: 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/internal/auth/basic.go b/pulsar/auth/basic.go
similarity index 100%
rename from pulsar/internal/auth/basic.go
rename to pulsar/auth/basic.go
diff --git a/pulsar/internal/auth/basic_test.go b/pulsar/auth/basic_test.go
similarity index 100%
rename from pulsar/internal/auth/basic_test.go
rename to pulsar/auth/basic_test.go
diff --git a/pulsar/internal/auth/disabled.go b/pulsar/auth/disabled.go
similarity index 100%
rename from pulsar/internal/auth/disabled.go
rename to pulsar/auth/disabled.go
diff --git a/pulsar/internal/auth/oauth2.go b/pulsar/auth/oauth2.go
similarity index 100%
rename from pulsar/internal/auth/oauth2.go
rename to pulsar/auth/oauth2.go
diff --git a/pulsar/internal/auth/oauth2_test.go b/pulsar/auth/oauth2_test.go
similarity index 100%
rename from pulsar/internal/auth/oauth2_test.go
rename to pulsar/auth/oauth2_test.go
diff --git a/pulsar/internal/auth/provider.go b/pulsar/auth/provider.go
similarity index 100%
rename from pulsar/internal/auth/provider.go
rename to pulsar/auth/provider.go
diff --git a/pulsar/internal/auth/tls.go b/pulsar/auth/tls.go
similarity index 100%
rename from pulsar/internal/auth/tls.go
rename to pulsar/auth/tls.go
diff --git a/pulsar/internal/auth/token.go b/pulsar/auth/token.go
similarity index 100%
rename from pulsar/internal/auth/token.go
rename to pulsar/auth/token.go
diff --git a/pulsar/client.go b/pulsar/client.go
index 29a582f..d75c977 100644
--- a/pulsar/client.go
+++ b/pulsar/client.go
@@ -21,7 +21,7 @@ import (
 	"crypto/tls"
 	"time"
 
-	"github.com/apache/pulsar-client-go/pulsar/internal/auth"
+	"github.com/apache/pulsar-client-go/pulsar/auth"
 	"github.com/apache/pulsar-client-go/pulsar/log"
 	"github.com/prometheus/client_golang/prometheus"
 )
diff --git a/pulsar/client_impl.go b/pulsar/client_impl.go
index 7cac097..e794e3d 100644
--- a/pulsar/client_impl.go
+++ b/pulsar/client_impl.go
@@ -22,12 +22,11 @@ import (
 	"net/url"
 	"time"
 
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/sirupsen/logrus"
-
+	"github.com/apache/pulsar-client-go/pulsar/auth"
 	"github.com/apache/pulsar-client-go/pulsar/internal"
-	"github.com/apache/pulsar-client-go/pulsar/internal/auth"
 	"github.com/apache/pulsar-client-go/pulsar/log"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/sirupsen/logrus"
 )
 
 const (
diff --git a/pulsar/consumer.go b/pulsar/consumer.go
index 0a515e0..c9f5e89 100644
--- a/pulsar/consumer.go
+++ b/pulsar/consumer.go
@@ -234,9 +234,20 @@ type Consumer interface {
 	// AckID the consumption of a single message, identified by its MessageID
 	AckID(MessageID) error
 
+	// AckCumulative the reception of all the messages in the stream up to (and including)
+	// the provided message.
+	AckCumulative(msg Message) error
+
+	// AckIDCumulative the reception of all the messages in the stream up to (and including)
+	// the provided message, identified by its MessageID
+	AckIDCumulative(msgID MessageID) error
+
 	// ReconsumeLater mark a message for redelivery after custom delay
 	ReconsumeLater(msg Message, delay time.Duration)
 
+	// ReconsumeLaterWithCustomProperties mark a message for redelivery after custom delay with custom properties
+	ReconsumeLaterWithCustomProperties(msg Message, customProperties map[string]string, delay time.Duration)
+
 	// Nack acknowledges the failure to process a single message.
 	//
 	// When a message is "negatively acked" it will be marked for redelivery after
@@ -267,9 +278,6 @@ type Consumer interface {
 
 	// SeekByTime resets the subscription associated with this consumer to a specific message publish time.
 	//
-	// Note: this operation can only be done on non-partitioned topics. For these, one can rather perform the seek() on
-	// the individual partitions.
-	//
 	// @param time
 	//            the message publish time when to reposition the subscription
 	//
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
index e5290a1..545b57c 100644
--- a/pulsar/consumer_impl.go
+++ b/pulsar/consumer_impl.go
@@ -38,6 +38,8 @@ type acker interface {
 	// AckID does not handle errors returned by the Broker side, so no need to wait for doneCh to finish.
 	AckID(id MessageID) error
 	AckIDWithResponse(id MessageID) error
+	AckIDCumulative(msgID MessageID) error
+	AckIDWithResponseCumulative(msgID MessageID) error
 	NackID(id MessageID)
 	NackMsg(msg Message)
 }
@@ -496,8 +498,34 @@ func (c *consumer) AckID(msgID MessageID) error {
 	return c.consumers[msgID.PartitionIdx()].AckID(msgID)
 }
 
+// AckCumulative the reception of all the messages in the stream up to (and including)
+// the provided message, identified by its MessageID
+func (c *consumer) AckCumulative(msg Message) error {
+	return c.AckIDCumulative(msg.ID())
+}
+
+// AckIDCumulative the reception of all the messages in the stream up to (and including)
+// the provided message, identified by its MessageID
+func (c *consumer) AckIDCumulative(msgID MessageID) error {
+	if err := c.checkMsgIDPartition(msgID); err != nil {
+		return err
+	}
+
+	if c.options.AckWithResponse {
+		return c.consumers[msgID.PartitionIdx()].AckIDWithResponseCumulative(msgID)
+	}
+
+	return c.consumers[msgID.PartitionIdx()].AckIDCumulative(msgID)
+}
+
 // ReconsumeLater mark a message for redelivery after custom delay
 func (c *consumer) ReconsumeLater(msg Message, delay time.Duration) {
+	c.ReconsumeLaterWithCustomProperties(msg, map[string]string{}, delay)
+}
+
+// ReconsumeLaterWithCustomProperties mark a message for redelivery after custom delay with custom properties
+func (c *consumer) ReconsumeLaterWithCustomProperties(msg Message, customProperties map[string]string,
+	delay time.Duration) {
 	if delay < 0 {
 		delay = 0
 	}
@@ -510,6 +538,10 @@ func (c *consumer) ReconsumeLater(msg Message, delay time.Duration) {
 		props[k] = v
 	}
 
+	for k, v := range customProperties {
+		props[k] = v
+	}
+
 	reconsumeTimes := 1
 	if s, ok := props[SysPropertyReconsumeTimes]; ok {
 		reconsumeTimes, _ = strconv.Atoi(s)
diff --git a/pulsar/consumer_multitopic.go b/pulsar/consumer_multitopic.go
index d32ce31..452915a 100644
--- a/pulsar/consumer_multitopic.go
+++ b/pulsar/consumer_multitopic.go
@@ -143,7 +143,39 @@ func (c *multiTopicConsumer) AckID(msgID MessageID) error {
 	return mid.consumer.AckID(msgID)
 }
 
+// AckCumulative the reception of all the messages in the stream up to (and including)
+// the provided message
+func (c *multiTopicConsumer) AckCumulative(msg Message) error {
+	return c.AckIDCumulative(msg.ID())
+}
+
+// AckIDCumulative the reception of all the messages in the stream up to (and including)
+// the provided message, identified by its MessageID
+func (c *multiTopicConsumer) AckIDCumulative(msgID MessageID) error {
+	mid, ok := toTrackingMessageID(msgID)
+	if !ok {
+		c.log.Warnf("invalid message id type %T", msgID)
+		return errors.New("invalid message id type in multi_consumer")
+	}
+
+	if mid.consumer == nil {
+		c.log.Warnf("unable to ack messageID=%+v can not determine topic", msgID)
+		return errors.New("unable to ack message because consumer is nil")
+	}
+
+	if c.options.AckWithResponse {
+		return mid.consumer.AckIDWithResponseCumulative(msgID)
+	}
+
+	return mid.consumer.AckIDCumulative(msgID)
+}
+
 func (c *multiTopicConsumer) ReconsumeLater(msg Message, delay time.Duration) {
+	c.ReconsumeLaterWithCustomProperties(msg, map[string]string{}, delay)
+}
+
+func (c *multiTopicConsumer) ReconsumeLaterWithCustomProperties(msg Message, customProperties map[string]string,
+	delay time.Duration) {
 	names, err := validateTopicNames(msg.Topic())
 	if err != nil {
 		c.log.Errorf("validate msg topic %q failed: %v", msg.Topic(), err)
@@ -165,7 +197,7 @@ func (c *multiTopicConsumer) ReconsumeLater(msg Message, delay time.Duration) {
 			return
 		}
 	}
-	consumer.ReconsumeLater(msg, delay)
+	consumer.ReconsumeLaterWithCustomProperties(msg, customProperties, delay)
 }
 
 func (c *multiTopicConsumer) Nack(msg Message) {
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index 4215b69..7f5b6df 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -451,6 +451,7 @@ func (pc *partitionConsumer) AckIDWithResponse(msgID MessageID) error {
 
 	ackReq := new(ackRequest)
 	ackReq.doneCh = make(chan struct{})
+	ackReq.ackType = individualAck
 	if !trackingID.Undefined() && trackingID.ack() {
 		pc.metrics.AcksCounter.Inc()
 		pc.metrics.ProcessingTime.Observe(float64(time.Now().UnixNano()-trackingID.receivedTime.UnixNano()) / 1.0e9)
@@ -483,6 +484,7 @@ func (pc *partitionConsumer) AckID(msgID MessageID) error {
 
 	ackReq := new(ackRequest)
 	ackReq.doneCh = make(chan struct{})
+	ackReq.ackType = individualAck
 	if !trackingID.Undefined() && trackingID.ack() {
 		pc.metrics.AcksCounter.Inc()
 		pc.metrics.ProcessingTime.Observe(float64(time.Now().UnixNano()-trackingID.receivedTime.UnixNano()) / 1.0e9)
@@ -497,6 +499,62 @@ func (pc *partitionConsumer) AckID(msgID MessageID) error {
 	return nil
 }
 
+func (pc *partitionConsumer) AckIDCumulative(msgID MessageID) error {
+	return pc.internalAckIDCumulative(msgID, false)
+}
+
+func (pc *partitionConsumer) AckIDWithResponseCumulative(msgID MessageID) error {
+	return pc.internalAckIDCumulative(msgID, true)
+}
+
+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")
+	}
+
+	// chunk message id will be converted to tracking message id
+	trackingID, ok := toTrackingMessageID(msgID)
+	if !ok {
+		return errors.New("failed to convert trackingMessageID")
+	}
+	if trackingID.Undefined() {
+		return nil
+	}
+
+	ackReq := new(ackRequest)
+	ackReq.doneCh = make(chan struct{})
+	ackReq.ackType = cumulativeAck
+	if trackingID.ackCumulative() {
+		ackReq.msgID = trackingID
+	} else if !trackingID.tracker.hasPrevBatchAcked() {
+		// get previous batch message id
+		ackReq.msgID = trackingID.prev()
+		trackingID.tracker.setPrevBatchAcked()
+	} else {
+		// waiting for all the msgs are acked in this batch
+		return nil
+	}
+
+	pc.metrics.AcksCounter.Inc()
+	pc.metrics.ProcessingTime.Observe(float64(time.Now().UnixNano()-trackingID.receivedTime.UnixNano()) / 1.0e9)
+	// send ack request to eventsCh
+	pc.eventsCh <- ackReq
+
+	if withResponse {
+		// wait for the request to complete if withResponse set true
+		<-ackReq.doneCh
+	}
+
+	pc.options.interceptors.OnAcknowledge(pc.parentConsumer, msgID)
+
+	if cmid, ok := toChunkedMessageID(msgID); ok {
+		pc.unAckChunksTracker.remove(cmid)
+	}
+
+	return nil
+}
+
 func (pc *partitionConsumer) NackID(msgID MessageID) {
 	if cmid, ok := toChunkedMessageID(msgID); ok {
 		pc.unAckChunksTracker.nack(cmid)
@@ -711,7 +769,13 @@ func (pc *partitionConsumer) internalAck(req *ackRequest) {
 	cmdAck := &pb.CommandAck{
 		ConsumerId: proto.Uint64(pc.consumerID),
 		MessageId:  messageIDs,
-		AckType:    pb.CommandAck_Individual.Enum(),
+	}
+
+	switch req.ackType {
+	case individualAck:
+		cmdAck.AckType = pb.CommandAck_Individual.Enum()
+	case cumulativeAck:
+		cmdAck.AckType = pb.CommandAck_Cumulative.Enum()
 	}
 
 	if pc.options.ackWithResponse {
@@ -1189,10 +1253,16 @@ func (pc *partitionConsumer) dispatcher() {
 	}
 }
 
+const (
+	individualAck = iota
+	cumulativeAck
+)
+
 type ackRequest struct {
-	doneCh chan struct{}
-	msgID  trackingMessageID
-	err    error
+	doneCh  chan struct{}
+	msgID   trackingMessageID
+	ackType int
+	err     error
 }
 
 type unsubscribeRequest struct {
diff --git a/pulsar/consumer_regex.go b/pulsar/consumer_regex.go
index 55f3d7a..d890c67 100644
--- a/pulsar/consumer_regex.go
+++ b/pulsar/consumer_regex.go
@@ -167,6 +167,11 @@ func (c *regexConsumer) ReconsumeLater(msg Message, delay time.Duration) {
 	c.log.Warnf("regexp consumer not support ReconsumeLater yet.")
 }
 
+func (c *regexConsumer) ReconsumeLaterWithCustomProperties(msg Message, customProperties map[string]string,
+	delay time.Duration) {
+	c.log.Warnf("regexp consumer not support ReconsumeLaterWithCustomProperties yet.")
+}
+
 // AckID the consumption of a single message, identified by its MessageID
 func (c *regexConsumer) AckID(msgID MessageID) error {
 	mid, ok := toTrackingMessageID(msgID)
@@ -187,6 +192,33 @@ func (c *regexConsumer) AckID(msgID MessageID) error {
 	return mid.consumer.AckID(msgID)
 }
 
+// AckCumulative the reception of all the messages in the stream up to (and including)
+// the provided message.
+func (c *regexConsumer) AckCumulative(msg Message) error {
+	return c.AckIDCumulative(msg.ID())
+}
+
+// AckIDCumulative the reception of all the messages in the stream up to (and including)
+// the provided message, identified by its MessageID
+func (c *regexConsumer) AckIDCumulative(msgID MessageID) error {
+	mid, ok := toTrackingMessageID(msgID)
+	if !ok {
+		c.log.Warnf("invalid message id type %T", msgID)
+		return errors.New("invalid message id type")
+	}
+
+	if mid.consumer == nil {
+		c.log.Warnf("unable to ack messageID=%+v can not determine topic", msgID)
+		return errors.New("unable to ack message because consumer is nil")
+	}
+
+	if c.options.AckWithResponse {
+		return mid.consumer.AckIDWithResponseCumulative(msgID)
+	}
+
+	return mid.consumer.AckIDCumulative(msgID)
+}
+
 func (c *regexConsumer) Nack(msg Message) {
 	if c.options.EnableDefaultNackBackoffPolicy || c.options.NackBackoffPolicy != nil {
 		msgID := msg.ID()
diff --git a/pulsar/impl_message.go b/pulsar/impl_message.go
index 3a7feee..d863da9 100644
--- a/pulsar/impl_message.go
+++ b/pulsar/impl_message.go
@@ -24,6 +24,7 @@ import (
 	"math/big"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"google.golang.org/protobuf/proto"
@@ -107,6 +108,25 @@ func (id trackingMessageID) ack() bool {
 	return true
 }
 
+func (id trackingMessageID) ackCumulative() bool {
+	if id.tracker != nil && id.batchIdx > -1 {
+		return id.tracker.ackCumulative(int(id.batchIdx))
+	}
+	return true
+}
+
+func (id trackingMessageID) prev() trackingMessageID {
+	return trackingMessageID{
+		messageID: messageID{
+			ledgerID:     id.ledgerID,
+			entryID:      id.entryID - 1,
+			partitionIdx: id.partitionIdx,
+		},
+		tracker:  id.tracker,
+		consumer: id.consumer,
+	}
+}
+
 func (id messageID) isEntryIDValid() bool {
 	return id.entryID >= 0
 }
@@ -367,8 +387,9 @@ func newAckTracker(size int) *ackTracker {
 
 type ackTracker struct {
 	sync.Mutex
-	size     int
-	batchIDs *big.Int
+	size           int
+	batchIDs       *big.Int
+	prevBatchAcked uint32
 }
 
 func (t *ackTracker) ack(batchID int) bool {
@@ -381,6 +402,25 @@ func (t *ackTracker) ack(batchID int) bool {
 	return len(t.batchIDs.Bits()) == 0
 }
 
+func (t *ackTracker) ackCumulative(batchID int) bool {
+	if batchID < 0 {
+		return true
+	}
+	mask := big.NewInt(-1)
+	t.Lock()
+	defer t.Unlock()
+	t.batchIDs.And(t.batchIDs, mask.Lsh(mask, uint(batchID+1)))
+	return len(t.batchIDs.Bits()) == 0
+}
+
+func (t *ackTracker) hasPrevBatchAcked() bool {
+	return atomic.LoadUint32(&t.prevBatchAcked) == 1
+}
+
+func (t *ackTracker) setPrevBatchAcked() {
+	atomic.StoreUint32(&t.prevBatchAcked, 1)
+}
+
 func (t *ackTracker) completed() bool {
 	t.Lock()
 	defer t.Unlock()
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index 00a9521..be95240 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -29,9 +29,10 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/apache/pulsar-client-go/pulsar/auth"
+
 	"google.golang.org/protobuf/proto"
 
-	"github.com/apache/pulsar-client-go/pulsar/internal/auth"
 	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
 	"github.com/apache/pulsar-client-go/pulsar/log"
 
diff --git a/pulsar/internal/connection_pool.go b/pulsar/internal/connection_pool.go
index cb172a1..4f6d656 100644
--- a/pulsar/internal/connection_pool.go
+++ b/pulsar/internal/connection_pool.go
@@ -24,7 +24,8 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/apache/pulsar-client-go/pulsar/internal/auth"
+	"github.com/apache/pulsar-client-go/pulsar/auth"
+
 	"github.com/apache/pulsar-client-go/pulsar/log"
 )
 
diff --git a/pulsar/internal/http_client.go b/pulsar/internal/http_client.go
index 7abf70d..672a65e 100644
--- a/pulsar/internal/http_client.go
+++ b/pulsar/internal/http_client.go
@@ -29,7 +29,7 @@ import (
 	"path"
 	"time"
 
-	"github.com/apache/pulsar-client-go/pulsar/internal/auth"
+	"github.com/apache/pulsar-client-go/pulsar/auth"
 
 	"github.com/apache/pulsar-client-go/pulsar/log"
 
diff --git a/pulsar/table_view_impl.go b/pulsar/table_view_impl.go
index 56b9fcb..85499f1 100644
--- a/pulsar/table_view_impl.go
+++ b/pulsar/table_view_impl.go
@@ -245,11 +245,17 @@ func (tv *TableViewImpl) handleMessage(msg Message) {
 	tv.dataMu.Lock()
 	defer tv.dataMu.Unlock()
 
-	payload := reflect.Indirect(reflect.New(tv.options.SchemaValueType)).Interface()
-	if err := msg.GetSchemaValue(&payload); err != nil {
-		tv.logger.Errorf("msg.GetSchemaValue() failed with %w; msg is %v", msg, err)
+	var payload interface{}
+	if len(msg.Payload()) == 0 {
+		delete(tv.data, msg.Key())
+	} else {
+		payload = reflect.Indirect(reflect.New(tv.options.SchemaValueType)).Interface()
+		if err := msg.GetSchemaValue(&payload); err != nil {
+			tv.logger.Errorf("msg.GetSchemaValue() failed with %w; msg is %v", msg, err)
+		}
+		tv.data[msg.Key()] = payload
 	}
-	tv.data[msg.Key()] = payload
+
 	for _, listener := range tv.listeners {
 		if err := listener(msg.Key(), payload); err != nil {
 			tv.logger.Errorf("table view listener failed for %v: %w", msg, err)


Test patch:
diff --git a/pulsar/internal/auth/athenz.go b/pulsar/auth/athenz.go
similarity index 100%
rename from pulsar/internal/auth/athenz.go
rename to pulsar/auth/athenz.go
diff --git a/pulsar/internal/auth/athenz_test.go b/pulsar/auth/athenz_test.go
similarity index 96%
rename from pulsar/internal/auth/athenz_test.go
rename to pulsar/auth/athenz_test.go
index 2556e93..97cc0ae 100644
--- a/pulsar/internal/auth/athenz_test.go
+++ b/pulsar/auth/athenz_test.go
@@ -27,10 +27,10 @@ import (
 	zms "github.com/AthenZ/athenz/libs/go/zmssvctoken"
 	zts "github.com/AthenZ/athenz/libs/go/ztsroletoken"
 	"github.com/stretchr/testify/assert"
-	mock "github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/mock"
 )
 
-const tlsClientKeyPath = "../../../integration-tests/certs/client-key.pem"
+const tlsClientKeyPath = "../../integration-tests/certs/client-key.pem"
 
 type MockTokenBuilder struct {
 	mock.Mock
diff --git a/pulsar/client_impl_test.go b/pulsar/client_impl_test.go
index c98a542..6ead057 100644
--- a/pulsar/client_impl_test.go
+++ b/pulsar/client_impl_test.go
@@ -28,12 +28,10 @@ import (
 	"testing"
 	"time"
 
-	"github.com/stretchr/testify/require"
-
+	"github.com/apache/pulsar-client-go/pulsar/auth"
 	"github.com/apache/pulsar-client-go/pulsar/internal"
-
-	"github.com/apache/pulsar-client-go/pulsar/internal/auth"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestClient(t *testing.T) {
diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
index 355e33b..11e72a7 100644
--- a/pulsar/consumer_test.go
+++ b/pulsar/consumer_test.go
@@ -867,6 +867,193 @@ func TestConsumerAck(t *testing.T) {
 	}
 }
 
+func TestConsumerNoBatchCumulativeAck(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topicName := newTopicName()
+	ctx := context.Background()
+
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic: topicName,
+		// disable batching
+		DisableBatching: true,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:            topicName,
+		SubscriptionName: "sub-1",
+		Type:             Exclusive,
+	})
+	assert.Nil(t, err)
+
+	const N = 100
+
+	for i := 0; i < N; i++ {
+		if _, err := producer.Send(ctx, &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("msg-content-%d", i)),
+		}); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	for i := 0; i < N; i++ {
+		msg, err := consumer.Receive(ctx)
+		assert.Nil(t, err)
+		assert.Equal(t, fmt.Sprintf("msg-content-%d", i), string(msg.Payload()))
+
+		if i == N/2-1 {
+			// cumulative acks the first half of messages
+			consumer.AckCumulative(msg)
+		}
+	}
+
+	consumer.Close()
+
+	// Subscribe again
+	consumer, err = client.Subscribe(ConsumerOptions{
+		Topic:            topicName,
+		SubscriptionName: "sub-1",
+		Type:             Exclusive,
+	})
+	assert.Nil(t, err)
+	defer consumer.Close()
+
+	// We should only receive the 2nd half of messages
+	for i := N / 2; i < N; i++ {
+		msg, err := consumer.Receive(ctx)
+		assert.Nil(t, err)
+		assert.Equal(t, fmt.Sprintf("msg-content-%d", i), string(msg.Payload()))
+
+		consumer.Ack(msg)
+	}
+}
+
+func TestConsumerBatchCumulativeAck(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topicName := newTopicName()
+	ctx := context.Background()
+
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic: topicName,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	c1, err := client.Subscribe(ConsumerOptions{
+		Topic:            topicName,
+		SubscriptionName: "sub-1",
+		Type:             Exclusive,
+	})
+	assert.Nil(t, err)
+
+	// c2 is used to test if previous batch can be acked
+	// when cumulative ack the next batch message id
+	c2, err := client.Subscribe(ConsumerOptions{
+		Topic:            topicName,
+		SubscriptionName: "sub-2",
+		Type:             Exclusive,
+	})
+	assert.Nil(t, err)
+
+	const N = 100
+
+	// send a batch
+	wg := sync.WaitGroup{}
+	for i := 0; i < N; i++ {
+		wg.Add(1)
+		producer.SendAsync(ctx, &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("msg-content-%d", i))},
+			func(id MessageID, producerMessage *ProducerMessage, e error) {
+				assert.NoError(t, e)
+				wg.Done()
+			})
+	}
+	wg.Wait()
+
+	err = producer.Flush()
+	assert.NoError(t, err)
+
+	// send another batch
+	wg = sync.WaitGroup{}
+	for i := N; i < 2*N; i++ {
+		wg.Add(1)
+		producer.SendAsync(ctx, &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("msg-content-%d", i))},
+			func(id MessageID, producerMessage *ProducerMessage, e error) {
+				assert.NoError(t, e)
+				wg.Done()
+			})
+	}
+	wg.Wait()
+
+	for i := 0; i < 2*N; i++ {
+		msg, err := c1.Receive(ctx)
+		assert.Nil(t, err)
+		assert.Equal(t, fmt.Sprintf("msg-content-%d", i), string(msg.Payload()))
+
+		if i == N-1 {
+			// cumulative ack the first half of messages
+			c1.AckCumulative(msg)
+		} else if i == N {
+			// the N+1 msg is in the second batch
+			// cumulative ack it to test if the first batch can be acked
+			c2.AckCumulative(msg)
+		}
+	}
+
+	c1.Close()
+	c2.Close()
+
+	// Subscribe again
+	c1, err = client.Subscribe(ConsumerOptions{
+		Topic:            topicName,
+		SubscriptionName: "sub-1",
+		Type:             Exclusive,
+	})
+	assert.Nil(t, err)
+	defer c1.Close()
+
+	// Subscribe again
+	c2, err = client.Subscribe(ConsumerOptions{
+		Topic:            topicName,
+		SubscriptionName: "sub-2",
+		Type:             Exclusive,
+	})
+	assert.Nil(t, err)
+	defer c2.Close()
+
+	// We should only receive the 2nd half of messages
+	for i := N; i < 2*N; i++ {
+		msg, err := c1.Receive(ctx)
+		assert.Nil(t, err)
+		assert.Equal(t, fmt.Sprintf("msg-content-%d", i), string(msg.Payload()))
+
+		c1.Ack(msg)
+	}
+
+	// We should only receive the 2nd half of messages
+	for i := N; i < 2*N; i++ {
+		msg, err := c2.Receive(ctx)
+		assert.Nil(t, err)
+		assert.Equal(t, fmt.Sprintf("msg-content-%d", i), string(msg.Payload()))
+
+		c2.Ack(msg)
+	}
+}
+
 func TestConsumerNack(t *testing.T) {
 	client, err := NewClient(ClientOptions{
 		URL: lookupURL,
@@ -1538,6 +1725,120 @@ func TestRLQ(t *testing.T) {
 	assert.Nil(t, checkMsg)
 }
 
+func TestRLQWithCustomProperties(t *testing.T) {
+	topic := newTopicName()
+	testURL := adminURL + "/" + "admin/v2/persistent/public/default/" + topic + "/partitions"
+	makeHTTPCall(t, http.MethodPut, testURL, "3")
+
+	subName := fmt.Sprintf("sub01-%d", time.Now().Unix())
+	maxRedeliveries := 2
+	N := 100
+	ctx := context.Background()
+
+	client, err := NewClient(ClientOptions{URL: lookupURL})
+	assert.Nil(t, err)
+	defer client.Close()
+
+	// 1. Pre-produce N messages
+	producer, err := client.CreateProducer(ProducerOptions{Topic: topic})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	for i := 0; i < N; i++ {
+		_, err = producer.Send(ctx, &ProducerMessage{Payload: []byte(fmt.Sprintf("MESSAGE_%d", i))})
+		assert.Nil(t, err)
+	}
+
+	// 2. Create consumer on the Retry Topic to reconsume N messages (maxRedeliveries+1) times
+	rlqConsumer, err := client.Subscribe(ConsumerOptions{
+		Topic:                       topic,
+		SubscriptionName:            subName,
+		Type:                        Shared,
+		SubscriptionInitialPosition: SubscriptionPositionEarliest,
+		DLQ: &DLQPolicy{
+			MaxDeliveries: uint32(maxRedeliveries),
+		},
+		RetryEnable:         true,
+		NackRedeliveryDelay: 1 * time.Second,
+	})
+	assert.Nil(t, err)
+	defer rlqConsumer.Close()
+
+	rlqReceived := 0
+	for rlqReceived < N*(maxRedeliveries+1) {
+		msg, err := rlqConsumer.Receive(ctx)
+		assert.Nil(t, err)
+
+		if msg.RedeliveryCount() > 0 {
+			msgProps := msg.Properties()
+
+			value, ok := msgProps["custom-key-1"]
+			assert.True(t, ok)
+			if ok {
+				assert.Equal(t, value, "custom-value-1")
+			}
+
+			rlqConsumer.ReconsumeLater(msg, 1*time.Second)
+		} else {
+			customProps := map[string]string{
+				"custom-key-1": "custom-val-1",
+			}
+			rlqConsumer.ReconsumeLaterWithCustomProperties(msg, customProps, 1*time.Second)
+		}
+
+		rlqReceived++
+	}
+	fmt.Println("retry consumed:", rlqReceived) // 300
+
+	// No more messages on the Retry Topic
+	rlqCtx, rlqCancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer rlqCancel()
+	msg, err := rlqConsumer.Receive(rlqCtx)
+	assert.Error(t, err)
+	assert.Nil(t, msg)
+
+	// 3. Create consumer on the DLQ topic to verify the routing
+	dlqConsumer, err := client.Subscribe(ConsumerOptions{
+		Topic:                       "persistent://public/default/" + topic + "-" + subName + "-DLQ",
+		SubscriptionName:            subName,
+		SubscriptionInitialPosition: SubscriptionPositionEarliest,
+	})
+	assert.Nil(t, err)
+	defer dlqConsumer.Close()
+
+	dlqReceived := 0
+	for dlqReceived < N {
+		msg, err := dlqConsumer.Receive(ctx)
+		assert.Nil(t, err)
+		dlqConsumer.Ack(msg)
+		dlqReceived++
+	}
+	fmt.Println("dlq received:", dlqReceived) // 100
+
+	// No more messages on the DLQ Topic
+	dlqCtx, dlqCancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer dlqCancel()
+	msg, err = dlqConsumer.Receive(dlqCtx)
+	assert.Error(t, err)
+	assert.Nil(t, msg)
+
+	// 4. No more messages for same subscription
+	checkConsumer, err := client.Subscribe(ConsumerOptions{
+		Topic:                       topic,
+		SubscriptionName:            subName,
+		Type:                        Shared,
+		SubscriptionInitialPosition: SubscriptionPositionEarliest,
+	})
+	assert.Nil(t, err)
+	defer checkConsumer.Close()
+
+	checkCtx, checkCancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer checkCancel()
+	checkMsg, err := checkConsumer.Receive(checkCtx)
+	assert.Error(t, err)
+	assert.Nil(t, checkMsg)
+}
+
 func TestAckWithResponse(t *testing.T) {
 	now := time.Now().Unix()
 	topic01 := fmt.Sprintf("persistent://public/default/topic-%d-01", now)
@@ -1573,6 +1874,43 @@ func TestAckWithResponse(t *testing.T) {
 	}
 }
 
+func TestCumulativeAckWithResponse(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:                        Exclusive,
+		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)
+	}
+
+	var msg Message
+	for i := 0; i < 10; i++ {
+		msg, err = consumer.Receive(ctx)
+		assert.Nil(t, err)
+	}
+
+	err = consumer.AckCumulative(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/impl_message_test.go b/pulsar/impl_message_test.go
index 3421ab4..89aab8a 100644
--- a/pulsar/impl_message_test.go
+++ b/pulsar/impl_message_test.go
@@ -80,6 +80,23 @@ func TestAckTracker(t *testing.T) {
 
 	}
 	assert.Equal(t, true, tracker.completed())
+
+	tracker = newAckTracker(1000)
+	for i := 0; i < 1000; i++ {
+		if i < 999 {
+			assert.False(t, tracker.ackCumulative(i))
+			assert.False(t, tracker.completed())
+		} else {
+			assert.True(t, tracker.ackCumulative(i))
+			assert.True(t, tracker.completed())
+		}
+	}
+
+	// test large number 1000 cumulative
+	tracker = newAckTracker(1000)
+
+	assert.Equal(t, true, tracker.ackCumulative(999))
+	assert.Equal(t, true, tracker.completed())
 }
 
 func TestAckingMessageIDBatchOne(t *testing.T) {
diff --git a/pulsar/internal/pulsartracing/consumer_interceptor_test.go b/pulsar/internal/pulsartracing/consumer_interceptor_test.go
index 9e70d8b..34e09d5 100644
--- a/pulsar/internal/pulsartracing/consumer_interceptor_test.go
+++ b/pulsar/internal/pulsartracing/consumer_interceptor_test.go
@@ -72,8 +72,20 @@ func (c *mockConsumer) AckID(msgID pulsar.MessageID) error {
 	return nil
 }
 
+func (c *mockConsumer) AckCumulative(msg pulsar.Message) error {
+	return nil
+}
+
+func (c *mockConsumer) AckIDCumulative(msgID pulsar.MessageID) error {
+	return nil
+}
+
 func (c *mockConsumer) ReconsumeLater(msg pulsar.Message, delay time.Duration) {}
 
+func (c *mockConsumer) ReconsumeLaterWithCustomProperties(msg pulsar.Message, customProperties map[string]string,
+	delay time.Duration) {
+}
+
 func (c *mockConsumer) Nack(msg pulsar.Message) {}
 
 func (c *mockConsumer) NackID(msgID pulsar.MessageID) {}
diff --git a/pulsar/table_view_test.go b/pulsar/table_view_test.go
index 829187b..d29b24d 100644
--- a/pulsar/table_view_test.go
+++ b/pulsar/table_view_test.go
@@ -25,6 +25,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestTableView(t *testing.T) {
@@ -78,3 +79,67 @@ func TestTableView(t *testing.T) {
 		assert.Equal(t, valuePrefix+k, *(v.(*string)))
 	}
 }
+
+func TestPublishNilValue(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.NoError(t, err)
+	defer client.Close()
+
+	topic := newTopicName()
+	schema := NewStringSchema(nil)
+
+	// create producer
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:  topic,
+		Schema: schema,
+	})
+	assert.NoError(t, err)
+	defer producer.Close()
+
+	// create table view
+	v := ""
+	tv, err := client.CreateTableView(TableViewOptions{
+		Topic:           topic,
+		Schema:          schema,
+		SchemaValueType: reflect.TypeOf(&v),
+	})
+	assert.NoError(t, err)
+	defer tv.Close()
+
+	_, err = producer.Send(context.Background(), &ProducerMessage{
+		Key:   "key-1",
+		Value: "value-1",
+	})
+	assert.NoError(t, err)
+
+	require.Eventually(t, func() bool {
+		return tv.Size() == 1
+	}, 5*time.Second, 100*time.Millisecond)
+
+	assert.Equal(t, *(tv.Get("key-1").(*string)), "value-1")
+
+	// send nil value
+	_, err = producer.Send(context.Background(), &ProducerMessage{
+		Key: "key-1",
+	})
+	assert.NoError(t, err)
+
+	require.Eventually(t, func() bool {
+		return tv.Size() == 0
+	}, 5*time.Second, 100*time.Millisecond)
+
+	_, err = producer.Send(context.Background(), &ProducerMessage{
+		Key:   "key-2",
+		Value: "value-2",
+	})
+	assert.NoError(t, err)
+
+	require.Eventually(t, func() bool {
+		return tv.Size() == 1
+	}, 5*time.Second, 100*time.Millisecond)
+
+	assert.Equal(t, *(tv.Get("key-2").(*string)), "value-2")
+}