[Reader] fix: deliver null-value tombstones instead of discarding them (#1482)

* fix(reader): deliver null-value tombstones instead of discarding them

A message published with MessageMetadata.null_value = true (the Pulsar
compaction tombstone convention) was conflated with a deserialization
failure in partitionConsumer.MessageReceived and silently discarded.
Because lastDequeuedMsg was never advanced past the tombstone,
hasMoreMessages kept returning true and Reader.Next blocked forever
when a tombstone was the last message on a topic.

Consumer: when the reader yields an empty payload and msgMeta or the
single-message metadata has null_value set, build a normal message with
payLoad == nil and take the usual dispatch path so lastDequeuedMsg
advances. Real corruption still routes through discardCorruptedMessage.

Producer: set MessageMetadata.null_value / SingleMessageMetadata.null_value
when both Value and Payload are nil, matching the Java client so
Go-produced tombstones carry the flag consumers need.

Message gains an IsNullValue() bool accessor so applications can tell
tombstones apart from empty payloads.

* fix: explicitly reseting err

Quoting Copilot's feedback:

When accepting a tombstone with err == internal.ErrEOM, the code keeps err
non-nil and continues. It works today because err is not used later, but it’s
easy to misread and could become a latent bug if future logic inspects/returns
err after this block. Consider explicitly setting err = nil in the
tombstone/ErrEOM acceptance path to make the intent unambiguous.

* fix: supress lint error

Author: jcmfernandes

pulsar/consumer_partition.go

@@ -1308,10 +1308,28 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 	)
 	for i := 0; i < numMsgs; i++ {
 		smm, payload, err := reader.ReadMessage()
-		if err != nil || payload == nil {
+		isNullValue := msgMeta.GetNullValue() || (smm != nil && smm.GetNullValue())
+		if err != nil {
+			// A null-value (tombstone) message has no payload bytes on the wire, so
+			// the non-batched reader returns ErrEOM. Accept it instead of discarding
+			// it as corrupted, matching the Java client's behavior for compaction
+			// tombstones.
+			if isNullValue && err == internal.ErrEOM {
+				payload = nil
+				// Explicit reset to make tombstone-acceptance
+				// intent unambiguous.
+				err = nil //nolint:ineffassign
+			} else {
+				pc.discardCorruptedMessage(pbMsgID, pb.CommandAck_BatchDeSerializeError)
+				return err
+			}
+		} else if payload == nil && !isNullValue {
 			pc.discardCorruptedMessage(pbMsgID, pb.CommandAck_BatchDeSerializeError)
 			return err
 		}
+		if isNullValue {
+			payload = nil
+		}
 		if ackSet != nil && !ackSet.Test(uint(i)) {
 			pc.log.Debugf("Ignoring message from %vth message, which has been acknowledged", i)
 			skippedMessages++
@@ -1396,6 +1414,7 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 				index:               messageIndex,
 				brokerPublishTime:   brokerPublishTime,
 				conn:                pc._getConn(),
+				isNullValue:         isNullValue,
 			}
 		} else {
 			msg = &message{
@@ -1417,6 +1436,7 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 				index:               messageIndex,
 				brokerPublishTime:   brokerPublishTime,
 				conn:                pc._getConn(),
+				isNullValue:         isNullValue,
 			}
 		}
 

pulsar/impl_message.go

@@ -316,6 +316,7 @@ type message struct {
 	index               *uint64
 	brokerPublishTime   *time.Time
 	conn                internal.Connection
+	isNullValue         bool
 }
 
 func (msg *message) Topic() string {
@@ -330,6 +331,10 @@ func (msg *message) Payload() []byte {
 	return msg.payLoad
 }
 
+func (msg *message) IsNullValue() bool {
+	return msg.isNullValue
+}
+
 func (msg *message) ID() MessageID {
 	return msg.msgID
 }

pulsar/internal/pulsartracing/message_carrier_util_test.go

@@ -80,6 +80,10 @@ func (msg *mockConsumerMessage) Payload() []byte {
 	return nil
 }
 
+func (msg *mockConsumerMessage) IsNullValue() bool {
+	return false
+}
+
 func (msg *mockConsumerMessage) ID() pulsar.MessageID {
 	return nil
 }

pulsar/message.go

@@ -90,6 +90,12 @@ type Message interface {
 	// Payload returns the payload of the message
 	Payload() []byte
 
+	// IsNullValue reports whether the message was published as a null-value
+	// (tombstone) message, i.e. with MessageMetadata.null_value set. For such
+	// messages Payload returns nil. Applications use this flag together with
+	// Pulsar topic compaction to mark a key as deleted.
+	IsNullValue() bool
+
 	// ID returns the unique message ID associated with this message.
 	// The message id can be used to univocally refer to a message without having the keep the entire payload in memory.
 	ID() MessageID

pulsar/negative_acks_tracker_test.go

@@ -197,6 +197,10 @@ func (msg *mockMessage1) Payload() []byte {
 	return nil
 }
 
+func (msg *mockMessage1) IsNullValue() bool {
+	return false
+}
+
 func (msg *mockMessage1) ID() MessageID {
 	return &messageID{
 		ledgerID: 1,
@@ -273,6 +277,10 @@ func (msg *mockMessage2) Payload() []byte {
 	return nil
 }
 
+func (msg *mockMessage2) IsNullValue() bool {
+	return false
+}
+
 func (msg *mockMessage2) ID() MessageID {
 	return &messageID{
 		ledgerID: 2,

pulsar/producer_partition.go

@@ -724,6 +724,10 @@ func (p *partitionProducer) genMetadata(msg *ProducerMessage,
 		UncompressedSize: proto.Uint32(uint32(uncompressedSize)),
 	}
 
+	if msg.Value == nil && msg.Payload == nil {
+		mm.NullValue = proto.Bool(true)
+	}
+
 	if !msg.EventTime.IsZero() {
 		mm.EventTime = proto.Uint64(internal.TimestampMillis(msg.EventTime))
 	}
@@ -771,6 +775,10 @@ func (p *partitionProducer) genSingleMessageMetadataInBatch(
 		PayloadSize: proto.Int32(int32(uncompressedSize)),
 	}
 
+	if msg.Value == nil && msg.Payload == nil {
+		smm.NullValue = proto.Bool(true)
+	}
+
 	if !msg.EventTime.IsZero() {
 		smm.EventTime = proto.Uint64(internal.TimestampMillis(msg.EventTime))
 	}

pulsar/reader_test.go

@@ -496,6 +496,70 @@ func TestReaderHasNext(t *testing.T) {
 	assert.Equal(t, 10, i)
 }
 
+// TestReaderNullValueTombstone verifies that a message published with a nil
+// Value / nil Payload (the Pulsar compaction tombstone convention) is delivered
+// to Reader.Next.
+func TestReaderNullValueTombstone(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topic := newTopicName()
+	ctx := context.Background()
+
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:           topic,
+		DisableBatching: true,
+	})
+	assert.Nil(t, err)
+
+	_, err = producer.Send(ctx, &ProducerMessage{
+		Key:     "k",
+		Payload: []byte("v1"),
+	})
+	assert.NoError(t, err)
+
+	tombstoneID, err := producer.Send(ctx, &ProducerMessage{
+		Key:     "k",
+		Payload: nil,
+	})
+	assert.NoError(t, err)
+	producer.Close()
+
+	reader, err := client.CreateReader(ReaderOptions{
+		Topic:                   topic,
+		StartMessageID:          EarliestMessageID(),
+		StartMessageIDInclusive: true,
+	})
+	assert.Nil(t, err)
+	defer reader.Close()
+
+	lastID, err := reader.GetLastMessageID()
+	assert.NoError(t, err)
+	assert.Equal(t, tombstoneID.LedgerID(), lastID.LedgerID())
+	assert.Equal(t, tombstoneID.EntryID(), lastID.EntryID())
+
+	firstCtx, firstCancel := context.WithTimeout(ctx, 5*time.Second)
+	defer firstCancel()
+	msg, err := reader.Next(firstCtx)
+	assert.NoError(t, err)
+	assert.Equal(t, []byte("v1"), msg.Payload())
+	assert.False(t, msg.IsNullValue())
+
+	secondCtx, secondCancel := context.WithTimeout(ctx, 5*time.Second)
+	defer secondCancel()
+	tombstone, err := reader.Next(secondCtx)
+	assert.NoError(t, err)
+	assert.True(t, tombstone.IsNullValue())
+	assert.Nil(t, tombstone.Payload())
+	assert.Equal(t, lastID.LedgerID(), tombstone.ID().LedgerID())
+	assert.Equal(t, lastID.EntryID(), tombstone.ID().EntryID())
+
+	assert.False(t, reader.HasNext())
+}
+
 type myMessageID struct {
 	data []byte
 }