Repository: apache/pulsar-client-go
Pull request #921 (2022-12-21 12:51:22)

[Issue 918] [Refactor] Remove the `clearMessageQueuesCh` in `partitionConsumer.dispatcher()`
Master Issue: #918 

### Motivation

The two chanel `clearMessageQueuesCh` and `clearQueueCb` just need to keep one.

For more details please check #918.

This PR does not ony aim to clean up, but also fix the potential bug in `clearMessageQueuesCh`.

For example, the `clearMessageQueuesCh` do the jod including clearing messageCh. But it may cause problem when `SeekByTime` invoked on partition topic.

https://github.com/apache/pulsar-client-go/blob/1d3499a18d526b4b1aef0bdbbc54ac812b8ae0c0/pulsar/consumer_impl.go#L614-L626

https://github.com/apache/pulsar-client-go/blob/1d3499a18d526b4b1aef0bdbbc54ac812b8ae0c0/pulsar/consumer_partition.go#L1168-L1175

When consume the partition topic, all the `partitionConsumer` share the same `messageCh`. After `SeekByTime` on partitioned topic, the messageCh may be cleared more than one time which will cause the messages losing. Suppose there is such a situation, partitionConsumer-1 has cleared its messageCh and queueCh. When partitionConsumer-2 do the clear job, it can also exec this logic.

https://github.com/apache/pulsar-client-go/blob/1d3499a18d526b4b1aef0bdbbc54ac812b8ae0c0/pulsar/consumer_partition.go#L1172-L1174

But `messageCh` is a share chan, partitionConsumer-1 may received new messages and put them to `messageCh` at this moment. There is such a possibility that partitionConsumer-2 **cleared the new messages** from `messageCh`.


### Modifications

- Remove the `clearMessageQueuesCh` in `partitionConsumer.dispatcher()`
- Modify the `clearQueueCb` in `partitionConsumer.dispatcher()`

### Verifying this change

- [x] Make sure that the change passes the CI checks.



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 0e89fc0..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
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
index e5290a1..e0120ad 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)
@@ -608,7 +640,16 @@ func (c *consumer) Seek(msgID MessageID) error {
 		return err
 	}
 
-	return c.consumers[msgID.PartitionIdx()].Seek(msgID)
+	if err := c.consumers[msgID.PartitionIdx()].Seek(msgID); err != nil {
+		return err
+	}
+
+	// clear messageCh
+	for len(c.messageCh) > 0 {
+		<-c.messageCh
+	}
+
+	return nil
 }
 
 func (c *consumer) SeekByTime(time time.Time) error {
@@ -622,6 +663,12 @@ func (c *consumer) SeekByTime(time time.Time) error {
 			errs = pkgerrors.Wrap(newError(SeekFailed, err.Error()), msg)
 		}
 	}
+
+	// clear messageCh
+	for len(c.messageCh) > 0 {
+		<-c.messageCh
+	}
+
 	return errs
 }
 
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..5ee09ab 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -145,15 +145,14 @@ type partitionConsumer struct {
 	// the size of the queue channel for buffering messages
 	queueSize       int32
 	queueCh         chan []*message
-	startMessageID  trackingMessageID
+	startMessageID  atomicMessageID
 	lastDequeuedMsg trackingMessageID
 
-	eventsCh             chan interface{}
-	connectedCh          chan struct{}
-	connectClosedCh      chan connectionClosed
-	closeCh              chan struct{}
-	clearQueueCh         chan func(id trackingMessageID)
-	clearMessageQueuesCh chan chan struct{}
+	eventsCh        chan interface{}
+	connectedCh     chan struct{}
+	connectClosedCh chan connectionClosed
+	closeCh         chan struct{}
+	clearQueueCh    chan func(id trackingMessageID)
 
 	nackTracker *negativeAcksTracker
 	dlq         *dlqRouter
@@ -212,6 +211,24 @@ func (p *availablePermits) reset() {
 	atomic.StoreInt32(&p.permits, 0)
 }
 
+// atomicMessageID is a wrapper for trackingMessageID to make get and set atomic
+type atomicMessageID struct {
+	msgID trackingMessageID
+	sync.RWMutex
+}
+
+func (a *atomicMessageID) get() trackingMessageID {
+	a.RLock()
+	defer a.RUnlock()
+	return a.msgID
+}
+
+func (a *atomicMessageID) set(msgID trackingMessageID) {
+	a.Lock()
+	defer a.Unlock()
+	a.msgID = msgID
+}
+
 type schemaInfoCache struct {
 	lock   sync.RWMutex
 	cache  map[string]Schema
@@ -277,13 +294,12 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 		eventsCh:             make(chan interface{}, 10),
 		queueSize:            int32(options.receiverQueueSize),
 		queueCh:              make(chan []*message, options.receiverQueueSize),
-		startMessageID:       options.startMessageID,
+		startMessageID:       atomicMessageID{msgID: options.startMessageID},
 		connectedCh:          make(chan struct{}),
 		messageCh:            messageCh,
 		connectClosedCh:      make(chan connectionClosed, 10),
 		closeCh:              make(chan struct{}),
 		clearQueueCh:         make(chan func(id trackingMessageID)),
-		clearMessageQueuesCh: make(chan chan struct{}),
 		compressionProviders: sync.Map{},
 		dlq:                  dlq,
 		metrics:              metrics,
@@ -324,14 +340,14 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 	pc.log.Info("Created consumer")
 	pc.setConsumerState(consumerReady)
 
-	if pc.options.startMessageIDInclusive && pc.startMessageID.equal(lastestMessageID.(messageID)) {
+	if pc.options.startMessageIDInclusive && pc.startMessageID.get().equal(lastestMessageID.(messageID)) {
 		msgID, err := pc.requestGetLastMessageID()
 		if err != nil {
 			pc.nackTracker.Close()
 			return nil, err
 		}
 		if msgID.entryID != noMessageEntry {
-			pc.startMessageID = msgID
+			pc.startMessageID.set(msgID)
 
 			// use the WithoutClear version because the dispatcher is not started yet
 			err = pc.requestSeekWithoutClear(msgID.messageID)
@@ -451,6 +467,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 +500,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 +515,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)
@@ -613,7 +687,7 @@ func (pc *partitionConsumer) requestSeek(msgID messageID) error {
 	if err := pc.requestSeekWithoutClear(msgID); err != nil {
 		return err
 	}
-	pc.clearMessageChannels()
+	pc.clearReceiverQueue()
 	return nil
 }
 
@@ -684,13 +758,7 @@ func (pc *partitionConsumer) internalSeekByTime(seek *seekByTimeRequest) {
 		seek.err = err
 		return
 	}
-	pc.clearMessageChannels()
-}
-
-func (pc *partitionConsumer) clearMessageChannels() {
-	doneCh := make(chan struct{})
-	pc.clearMessageQueuesCh <- doneCh
-	<-doneCh
+	pc.clearReceiverQueue()
 }
 
 func (pc *partitionConsumer) internalAck(req *ackRequest) {
@@ -711,7 +779,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 {
@@ -993,7 +1067,7 @@ func (pc *partitionConsumer) processMessageChunk(compressedPayload internal.Buff
 }
 
 func (pc *partitionConsumer) messageShouldBeDiscarded(msgID trackingMessageID) bool {
-	if pc.startMessageID.Undefined() {
+	if pc.startMessageID.get().Undefined() {
 		return false
 	}
 	// if we start at latest message, we should never discard
@@ -1002,11 +1076,11 @@ func (pc *partitionConsumer) messageShouldBeDiscarded(msgID trackingMessageID) b
 	}
 
 	if pc.options.startMessageIDInclusive {
-		return pc.startMessageID.greater(msgID.messageID)
+		return pc.startMessageID.get().greater(msgID.messageID)
 	}
 
 	// Non inclusive
-	return pc.startMessageID.greaterEqual(msgID.messageID)
+	return pc.startMessageID.get().greaterEqual(msgID.messageID)
 }
 
 // create EncryptionContext from message metadata
@@ -1154,45 +1228,33 @@ func (pc *partitionConsumer) dispatcher() {
 			go func() {
 				pc.queueCh <- nil
 			}()
+
 			for m := range pc.queueCh {
 				// the queue has been drained
 				if m == nil {
 					break
 				} else if nextMessageInQueue.Undefined() {
-					nextMessageInQueue = m[0].msgID.(trackingMessageID)
+					nextMessageInQueue, _ = toTrackingMessageID(m[0].msgID)
 				}
 			}
 
-			clearQueueCb(nextMessageInQueue)
-
-		case doneCh := <-pc.clearMessageQueuesCh:
-			for len(pc.queueCh) > 0 {
-				<-pc.queueCh
-			}
-			for len(pc.messageCh) > 0 {
-				<-pc.messageCh
-			}
 			messages = nil
 
-			// reset available permits
-			pc.availablePermits.reset()
-			initialPermits := uint32(pc.queueSize)
-
-			pc.log.Debugf("dispatcher requesting initial permits=%d", initialPermits)
-			// send initial permits
-			if err := pc.internalFlow(initialPermits); err != nil {
-				pc.log.WithError(err).Error("unable to send initial permits to broker")
-			}
-
-			close(doneCh)
+			clearQueueCb(nextMessageInQueue)
 		}
 	}
 }
 
+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 {
@@ -1420,10 +1482,10 @@ func (pc *partitionConsumer) grabConn() error {
 		KeySharedMeta:              keySharedMeta,
 	}
 
-	pc.startMessageID = pc.clearReceiverQueue()
+	pc.startMessageID.set(pc.clearReceiverQueue())
 	if pc.options.subscriptionMode != durable {
 		// For regular subscriptions the broker will determine the restarting point
-		cmdSubscribe.StartMessageId = convertToMessageIDData(pc.startMessageID)
+		cmdSubscribe.StartMessageId = convertToMessageIDData(pc.startMessageID.get())
 	}
 
 	if len(pc.options.metadata) > 0 {
@@ -1501,8 +1563,8 @@ func (pc *partitionConsumer) clearQueueAndGetNextMessage() trackingMessageID {
 func (pc *partitionConsumer) clearReceiverQueue() trackingMessageID {
 	nextMessageInQueue := pc.clearQueueAndGetNextMessage()
 
-	if pc.startMessageID.Undefined() {
-		return pc.startMessageID
+	if pc.startMessageID.get().Undefined() {
+		return pc.startMessageID.get()
 	}
 
 	if !nextMessageInQueue.Undefined() {
@@ -1513,7 +1575,7 @@ func (pc *partitionConsumer) clearReceiverQueue() trackingMessageID {
 		return pc.lastDequeuedMsg
 	} else {
 		// No message was received or dequeued by this consumer. Next message would still be the startMessageId
-		return pc.startMessageID
+		return pc.startMessageID.get()
 	}
 }
 
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/reader_impl.go b/pulsar/reader_impl.go
index 3931f36..079754b 100644
--- a/pulsar/reader_impl.go
+++ b/pulsar/reader_impl.go
@@ -185,11 +185,13 @@ func (r *reader) hasMoreMessages() bool {
 	}
 
 	if r.pc.options.startMessageIDInclusive {
-		return r.lastMessageInBroker.isEntryIDValid() && r.lastMessageInBroker.greaterEqual(r.pc.startMessageID.messageID)
+		return r.lastMessageInBroker.isEntryIDValid() &&
+			r.lastMessageInBroker.greaterEqual(r.pc.startMessageID.get().messageID)
 	}
 
 	// Non-inclusive
-	return r.lastMessageInBroker.isEntryIDValid() && r.lastMessageInBroker.greater(r.pc.startMessageID.messageID)
+	return r.lastMessageInBroker.isEntryIDValid() &&
+		r.lastMessageInBroker.greater(r.pc.startMessageID.get().messageID)
 }
 
 func (r *reader) Close() {
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")
+}