Add retention setting to tx_single_reads_writes scenario

harness/chaos/redpanda_static_cluster.py

@@ -218,8 +218,8 @@ def wait_alive(self, timeout_s=10):
     def brokers(self):
         return ",".join([x.ip+":9092" for x in self.hosts])
 
-    def create_topic(self, topic, replication, partitions, cleanup="delete"):
-        ssh("ubuntu@" + self.nodes[0].ip, "rpk", "topic", "create", "--brokers", self.brokers(), topic, "-r", replication, "-p", partitions, "-c", f"cleanup.policy={cleanup}")
+    def create_topic(self, topic, replication, partitions, cleanup="delete", retention_ms=604800000, segment_bytes=1073741824):
+        ssh("ubuntu@" + self.nodes[0].ip, "rpk", "topic", "create", "--brokers", self.brokers(), topic, "-r", replication, "-p", partitions, "-c", f"cleanup.policy={cleanup}", "-c", f"retention.ms={retention_ms}", "-c", f"segment.bytes={segment_bytes}")
 
     def reconfigure(self, leader, replicas, topic, partition=0, namespace="kafka"):
         payload = []

harness/chaos/scenarios/tx_single_topic_single_fault.py

@@ -95,8 +95,17 @@ def prepare_experiment(self, config, experiment_id):
         # waiting for the controller to be up before creating a topic
         self.redpanda_cluster.wait_leader("controller", namespace="redpanda", replication=len(self.redpanda_cluster.nodes), timeout_s=30)
 
+        retention_ms = 604800000
+        if self.config.get("retention_ms") != None:
+            retention_ms = self.config["retention_ms"]
+
+        segment_bytes = 1073741824
+        if self.config.get("segment_bytes") != None:
+            segment_bytes = self.config["segment_bytes"]
+
+
         logger.info(f"creating \"{self.topic}\" topic with replication factor {self.replication}")
-        self.redpanda_cluster.create_topic(self.topic, self.replication, 1)
+        self.redpanda_cluster.create_topic(self.topic, self.replication, 1, retention_ms=retention_ms, segment_bytes=segment_bytes)
         # waiting for the topic to come online
         self.redpanda_cluster.wait_leader(self.topic, replication=self.replication, timeout_s=20)
 

suites/test_suite_tx_reads_writes.json

@@ -17,6 +17,7 @@
         "tx_single_reads_writes/pause_tx_leader.json",
         "tx_single_reads_writes/reconfigure_313_data.json",
         "tx_single_reads_writes/reconfigure_313_tx.json",
+        "tx_single_reads_writes/retention.json",
         "tx_single_reads_writes/transfer_data_leadership.json",
         "tx_single_reads_writes/transfer_tx_leadership.json",
         "tx_single_reads_writes/recycle_all.json"

suites/tests/tx_single_reads_writes/retention.json

@@ -0,0 +1,24 @@
+{
+    "name": "tx-single-reads-writes / java / baseline",
+    "scenario": "tx_single_table_single_fault",
+    "topic": "topic1",
+    "replication": 3,
+    "retention_ms": 10,
+    "segment_bytes": 15000,
+    "workload": {
+        "name": "tx-single-reads-writes / java",
+        "checks": [
+            { "name": "consistency" },
+            { "name": "stat" }
+        ],
+        "settings": {
+            "reads": 3,
+            "writes": 2,
+            "retries": 5
+        }
+    },
+    "fault": null,
+    "checks": [
+        { "name": "redpanda_process_liveness" }
+    ]
+}