Repository: apache/pulsar-client-go
Pull request #930 (2022-12-30 08:31:36)

Handle nil value message correctly in table-view
### Motivation

Currently, when we publish a nil value message([tombstones](https://en.wikipedia.org/wiki/Tombstone_(data_store))) to a compacted topic used to `TableView`, the `TableView` will store a nil value in the map, but we should remove this key since this is a tombstone.

### Modifications

* Handle nil value message correctly in table-view
* Add unit test to verify.





Patch:
diff --git a/pulsar/consumer.go b/pulsar/consumer.go
index 0e89fc0..49a24d5 100644
--- a/pulsar/consumer.go
+++ b/pulsar/consumer.go
@@ -234,6 +234,14 @@ 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)
 
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
index e5290a1..11a3cf7 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,6 +498,26 @@ 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) {
 	if delay < 0 {
diff --git a/pulsar/consumer_multitopic.go b/pulsar/consumer_multitopic.go
index d32ce31..e653926 100644
--- a/pulsar/consumer_multitopic.go
+++ b/pulsar/consumer_multitopic.go
@@ -143,6 +143,33 @@ 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) {
 	names, err := validateTopicNames(msg.Topic())
 	if err != nil {
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..36d2245 100644
--- a/pulsar/consumer_regex.go
+++ b/pulsar/consumer_regex.go
@@ -187,6 +187,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/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/consumer_test.go b/pulsar/consumer_test.go
index 355e33b..891fcd2 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,
@@ -1573,6 +1760,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..a06e4ee 100644
--- a/pulsar/internal/pulsartracing/consumer_interceptor_test.go
+++ b/pulsar/internal/pulsartracing/consumer_interceptor_test.go
@@ -72,6 +72,14 @@ 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) Nack(msg pulsar.Message) {}
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")
+}