Support for tombstone null message

Author: Bhargavkonidena

include/pulsar/Message.h

@@ -159,6 +159,17 @@ class PULSAR_PUBLIC Message {
      */
     bool hasOrderingKey() const;
 
+    /**
+     * Check if the message has a null value.
+     *
+     * Messages with null values are used as tombstones on compacted topics
+     * to delete the message for a specific key.
+     *
+     * @return true if the message has a null value (tombstone)
+     *         false if the message has actual payload data
+     */
+    bool hasNullValue() const;
+
     /**
      * Get the UTC based timestamp in milliseconds referring to when the message was published by the client
      * producer

include/pulsar/MessageBuilder.h

@@ -156,6 +156,17 @@ class PULSAR_PUBLIC MessageBuilder {
      */
     MessageBuilder& disableReplication(bool flag);
 
+    /**
+     * Mark the message as having a null value.
+     *
+     * This is used for messages on compacted topics where a null value
+     * acts as a tombstone for a specific key, removing the message from
+     * the compacted view.
+     *
+     * @return the message builder instance
+     */
+    MessageBuilder& setNullValue();
+
     /**
      * create a empty message, with no properties or data
      *

include/pulsar/c/message.h

@@ -127,6 +127,15 @@ PULSAR_PUBLIC void pulsar_message_set_replication_clusters(pulsar_message_t *mes
  */
 PULSAR_PUBLIC void pulsar_message_disable_replication(pulsar_message_t *message, int flag);
 
+/**
+ * Mark the message as having a null value.
+ *
+ * This is used for messages on compacted topics where a null value
+ * acts as a tombstone for a specific key, removing the message from
+ * the compacted view.
+ */
+PULSAR_PUBLIC void pulsar_message_set_null_value(pulsar_message_t *message);
+
 /// Accessor for built messages
 
 /**
@@ -221,6 +230,16 @@ PULSAR_PUBLIC void pulsar_message_set_schema_version(pulsar_message_t *message,
  */
 PULSAR_PUBLIC const char *pulsar_message_get_producer_name(pulsar_message_t *message);
 
+/**
+ * Check if the message has a null value.
+ *
+ * Messages with null values are used as tombstones on compacted topics
+ * to delete the message for a specific key.
+ *
+ * @return 1 if the message has a null value, 0 otherwise
+ */
+PULSAR_PUBLIC int pulsar_message_has_null_value(pulsar_message_t *message);
+
 #ifdef __cplusplus
 }
 #endif

lib/Commands.cc

@@ -871,6 +871,10 @@ static std::pair<std::unique_ptr<char[]>, size_t> serializeSingleMessageMetadata
         metadata.set_sequence_id(msgMetadata.sequence_id());
     }
 
+    if (msgMetadata.null_value()) {
+        metadata.set_null_value(true);
+    }
+
     size_t size = metadata.ByteSizeLong();
     std::unique_ptr<char[]> data{new char[size]};
     metadata.SerializeToArray(data.get(), size);

lib/Message.cc

@@ -123,6 +123,12 @@ Message::Message(const MessageId& messageID, proto::BrokerEntryMetadata& brokerE
     } else {
         impl_->metadata.clear_sequence_id();
     }
+
+    if (singleMetadata.null_value()) {
+        impl_->metadata.set_null_value(true);
+    } else {
+        impl_->metadata.clear_null_value();
+    }
 }
 
 const MessageId& Message::getMessageId() const {
@@ -177,6 +183,13 @@ const std::string& Message::getOrderingKey() const {
     return impl_->getOrderingKey();
 }
 
+bool Message::hasNullValue() const {
+    if (impl_) {
+        return impl_->metadata.null_value();
+    }
+    return false;
+}
+
 const std::string& Message::getTopicName() const {
     if (!impl_) {
         return emptyString;

lib/MessageBuilder.cc

@@ -157,6 +157,12 @@ MessageBuilder& MessageBuilder::disableReplication(bool flag) {
     return *this;
 }
 
+MessageBuilder& MessageBuilder::setNullValue() {
+    checkMetadata();
+    impl_->metadata.set_null_value(true);
+    return *this;
+}
+
 const char* MessageBuilder::data() const {
     assert(impl_->payload.data());
     return impl_->payload.data();

lib/c/c_Message.cc

@@ -81,6 +81,8 @@ void pulsar_message_disable_replication(pulsar_message_t *message, int flag) {
     message->builder.disableReplication(flag);
 }
 
+void pulsar_message_set_null_value(pulsar_message_t *message) { message->builder.setNullValue(); }
+
 int pulsar_message_has_property(pulsar_message_t *message, const char *name) {
     return message->message.hasProperty(name);
 }
@@ -148,3 +150,5 @@ void pulsar_message_set_schema_version(pulsar_message_t *message, const char *sc
 const char *pulsar_message_get_producer_name(pulsar_message_t *message) {
     return message->message.getProducerName().c_str();
 }
+
+int pulsar_message_has_null_value(pulsar_message_t *message) { return message->message.hasNullValue(); }

tests/BatchMessageTest.cc

@@ -988,6 +988,36 @@ TEST(BatchMessageTest, testParseMessageBatchEntry) {
     }
 }
 
+TEST(BatchMessageTest, testParseMessageBatchEntryWithNullValue) {
+    std::vector<Message> msgs;
+    msgs.emplace_back(MessageBuilder().setPartitionKey("key1").setNullValue().build());
+    msgs.emplace_back(MessageBuilder().setContent("content2").setPartitionKey("key2").build());
+    msgs.emplace_back(MessageBuilder().setPartitionKey("key3").setNullValue().build());
+
+    SharedBuffer payload;
+    Commands::serializeSingleMessagesToBatchPayload(payload, msgs);
+    ASSERT_EQ(payload.writableBytes(), 0);
+
+    MessageBatch messageBatch;
+    auto fakeId = MessageIdBuilder().ledgerId(6000L).entryId(20L).partition(0).build();
+    messageBatch.withMessageId(fakeId).parseFrom(payload, static_cast<uint32_t>(msgs.size()));
+    const std::vector<Message>& messages = messageBatch.messages();
+
+    ASSERT_EQ(messages.size(), 3);
+
+    ASSERT_TRUE(messages[0].hasNullValue());
+    ASSERT_EQ(messages[0].getPartitionKey(), "key1");
+    ASSERT_EQ(messages[0].getLength(), 0);
+
+    ASSERT_FALSE(messages[1].hasNullValue());
+    ASSERT_EQ(messages[1].getPartitionKey(), "key2");
+    ASSERT_EQ(messages[1].getDataAsString(), "content2");
+
+    ASSERT_TRUE(messages[2].hasNullValue());
+    ASSERT_EQ(messages[2].getPartitionKey(), "key3");
+    ASSERT_EQ(messages[2].getLength(), 0);
+}
+
 TEST(BatchMessageTest, testSendCallback) {
     const std::string topicName = "persistent://public/default/BasicMessageTest-testSendCallback";
 

tests/MessageTest.cc

@@ -153,3 +153,46 @@ TEST(MessageTest, testGetTopicNameOnProducerMessage) {
     auto msg = MessageBuilder().setContent("test").build();
     ASSERT_TRUE(msg.getTopicName().empty());
 }
+
+TEST(MessageTest, testNullValueMessage) {
+    {
+        auto msg = MessageBuilder().setContent("test").build();
+        ASSERT_FALSE(msg.hasNullValue());
+    }
+
+    {
+        auto msg = MessageBuilder().setNullValue().setPartitionKey("key1").build();
+        ASSERT_TRUE(msg.hasNullValue());
+        ASSERT_EQ(msg.getLength(), 0);
+        ASSERT_EQ(msg.getPartitionKey(), "key1");
+    }
+
+    {
+        auto msg = MessageBuilder().setPartitionKey("key2").setNullValue().build();
+        ASSERT_TRUE(msg.hasNullValue());
+        ASSERT_EQ(msg.getPartitionKey(), "key2");
+    }
+}
+
+TEST(MessageTest, testEmptyMessage) {
+    auto msg = MessageBuilder().build();
+    ASSERT_FALSE(msg.hasNullValue());
+    ASSERT_EQ(msg.getLength(), 0);
+}
+
+TEST(MessageTest, testEmptyStringNotNullValue) {
+    // Empty string message - has content set to ""
+    auto emptyStringMsg = MessageBuilder().setContent("").build();
+    ASSERT_FALSE(emptyStringMsg.hasNullValue());
+    ASSERT_EQ(emptyStringMsg.getLength(), 0);
+    ASSERT_EQ(emptyStringMsg.getDataAsString(), "");
+
+    // Null value message - explicitly marked as null
+    auto nullValueMsg = MessageBuilder().setNullValue().setPartitionKey("key").build();
+    ASSERT_TRUE(nullValueMsg.hasNullValue());
+    ASSERT_EQ(nullValueMsg.getLength(), 0);
+
+    // Both have length 0, but they are semantically different
+    // Empty string: the value IS an empty string
+    // Null value: the value does not exist (tombstone for compaction)
+}