From 8d35402fba46cdbb8161114d453b8bd5a007ad94 Mon Sep 17 00:00:00 2001
From: Willem Kaufmann <willem.kaufmann@redpanda.com>
Date: Tue, 10 Sep 2024 16:15:21 -0400
Subject: [PATCH] `rptest`: add
 `RpkToolTest::test_produce_and_consume_tombstones`

---
 tests/rptest/tests/rpk_topic_test.py | 43 ++++++++++++++++++++++++++++
 1 file changed, 43 insertions(+)

diff --git a/tests/rptest/tests/rpk_topic_test.py b/tests/rptest/tests/rpk_topic_test.py
index a22b615a76ec..003c29f4e1b1 100644
--- a/tests/rptest/tests/rpk_topic_test.py
+++ b/tests/rptest/tests/rpk_topic_test.py
@@ -240,3 +240,46 @@ def cond():
 
         raise ducktape.errors.TimeoutError(
             f'Message in {topic} never appeared.')
+
+    @cluster(num_nodes=4)
+    def test_produce_and_consume_tombstones(self):
+        topic = 'rp_dt_test_produce_and_consume_tombstones'
+        self._rpk.create_topic(topic)
+
+        num_messages = 2
+
+        tombstone_key = 'ISTOMBSTONE'
+        tombstone_value = ''
+
+        # Producing a record with an empty value and -Z flag results in a tombstone.
+        self._rpk.produce(topic,
+                          tombstone_key,
+                          tombstone_value,
+                          tombstone=True)
+
+        not_tombstone_key = 'ISNOTTOMBSTONE'
+
+        # Producing a record with an empty value without the -Z flag results in an empty value.
+        self._rpk.produce(topic,
+                          not_tombstone_key,
+                          tombstone_value,
+                          tombstone=False)
+
+        c = RpkConsumer(self._ctx, self.redpanda, topic)
+        c.start()
+
+        def cond():
+            return len(c.messages) == num_messages
+
+        wait_until(cond,
+                   timeout_sec=30,
+                   backoff_sec=2,
+                   err_msg=f'Messages in {topic} never appeared.')
+
+        # Tombstone records do not have a value in the returned JSON
+        assert c.messages[0]['key'] == tombstone_key
+        assert 'value' not in c.messages[0]
+
+        # Records with an empty string have a value of `""` in the returned JSON
+        assert c.messages[1]['key'] == not_tombstone_key
+        assert c.messages[1]['value'] == ""