Handle nil value message correctly in table-view

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)

pulsar/table_view_test.go

@@ -78,3 +78,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)
+
+	for tv.Size() < 1 {
+		time.Sleep(time.Second * 1)
+	}
+
+	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)
+
+	for tv.Size() >= 1 {
+		time.Sleep(time.Second * 1)
+	}
+
+	_, err = producer.Send(context.Background(), &ProducerMessage{
+		Key:   "key-2",
+		Value: "value-2",
+	})
+	assert.NoError(t, err)
+
+	for tv.Size() < 1 {
+		time.Sleep(time.Second * 1)
+	}
+
+	assert.Equal(t, *(tv.Get("key-2").(*string)), "value-2")
+}