Merge pull request #28428 from Lazin/fix/timequery-test

[CORE-14434] rptest: Fix timequery test

Author: Lazin

tests/rptest/tests/timequery_test.py

@@ -48,8 +48,10 @@ class BaseTimeQuery:
     test_context: dict
     logger: Logger
 
+    base_ts = int(time.time() - 600) * 1000
+
     def _create_and_produce(
-        self, cluster, cloud_storage, local_retention, base_ts, record_size, msg_count
+        self, cluster, cloud_storage, local_retention, record_size, msg_count
     ):
         topic = TopicSpec(name="tqtopic", partition_count=1, replication_factor=3)
         self.client().create_topic(topic)
@@ -95,20 +97,19 @@ def _create_and_produce(
             # but small enough that we are getting plenty of distinct batches.
             batch_max_bytes=record_size * 10,
             # A totally arbitrary artificial timestamp base in milliseconds
-            fake_timestamp_ms=base_ts,
+            fake_timestamp_ms=self.base_ts,
         )
         producer.start()
         producer.wait()
 
         # We know timestamps, they are generated linearly from the
         # base we provided to kgo-verifier
-        timestamps = dict((i, base_ts + i) for i in range(0, msg_count))
+        timestamps = dict((i, self.base_ts + i) for i in range(0, msg_count))
         return topic, timestamps
 
     def _test_timequery(self, cluster, cloud_storage: bool, batch_cache: bool):
         total_segments = 12
         record_size = 1024
-        base_ts = 1664453149000
         msg_count = (self.log_segment_size * total_segments) // record_size
         local_retention = self.log_segment_size * 4
         kcat = KafkaCat(cluster)
@@ -118,13 +119,13 @@ def _test_timequery(self, cluster, cloud_storage: bool, batch_cache: bool):
             name="tq_empty_topic", partition_count=1, replication_factor=3
         )
         self.client().create_topic(empty_topic)
-        offset = kcat.query_offset(empty_topic.name, 0, base_ts)
+        offset = kcat.query_offset(empty_topic.name, 0, self.base_ts)
         self.logger.info(f"Time query returned offset {offset}")
         assert offset == -1, f"Expected -1, got {offset}"
 
         # Create a topic and produce a run of messages we will query.
         topic, timestamps = self._create_and_produce(
-            cluster, cloud_storage, local_retention, base_ts, record_size, msg_count
+            cluster, cloud_storage, local_retention, record_size, msg_count
         )
         for k, v in timestamps.items():
             self.logger.debug(f"  Offset {k} -> Timestamp {v}")
@@ -270,11 +271,10 @@ def _test_timequery_below_start_offset(self, cluster):
         total_segments = 3
         local_retain_segments = 1
         record_size = 1024
-        base_ts = 1664453149000
         msg_count = (self.log_segment_size * total_segments) // record_size
 
         topic, timestamps = self._create_and_produce(
-            cluster, False, local_retain_segments, base_ts, record_size, msg_count
+            cluster, False, local_retain_segments, record_size, msg_count
         )
 
         self.client().alter_topic_config(
@@ -295,7 +295,7 @@ def start_offset():
 
         kcat = KafkaCat(cluster)
         lwm_before = start_offset()
-        offset = kcat.query_offset(topic.name, 0, base_ts)
+        offset = kcat.query_offset(topic.name, 0, self.base_ts)
         lwm_after = start_offset()
 
         # When querying before the start of the log, we should get
@@ -416,11 +416,10 @@ def test_timequery_with_local_gc(self):
         self.set_up_cluster(cloud_storage=True, batch_cache=False, spillover=False)
         local_retention = self.log_segment_size * 4
         record_size = 1024
-        base_ts = 1664453149000
         msg_count = (self.log_segment_size * total_segments) // record_size
 
         topic, timestamps = self._create_and_produce(
-            self.redpanda, True, local_retention, base_ts, record_size, msg_count
+            self.redpanda, True, local_retention, record_size, msg_count
         )
 
         # While waiting for local GC to occur, run several concurrent
@@ -494,11 +493,10 @@ def test_timequery_with_trim_prefix(self, cloud_storage: bool, spillover: bool):
         )
         total_segments = 12
         record_size = 1024
-        base_ts = 1664453149000
         msg_count = (self.log_segment_size * total_segments) // record_size
         local_retention = self.log_segment_size * 4
         topic, timestamps = self._create_and_produce(
-            self.redpanda, True, local_retention, base_ts, record_size, msg_count
+            self.redpanda, True, local_retention, record_size, msg_count
         )
 
         # Confirm messages written
@@ -554,12 +552,11 @@ def test_timequery_with_spillover_gc_delayed(self):
         self.set_up_cluster(cloud_storage=True, batch_cache=False, spillover=True)
         total_segments = 16
         record_size = 1024
-        base_ts = 1664453149000
         msg_count = (self.log_segment_size * total_segments) // record_size
         local_retention = self.log_segment_size * 4
         topic_retention = self.log_segment_size * 8
         topic, timestamps = self._create_and_produce(
-            self.redpanda, True, local_retention, base_ts, record_size, msg_count
+            self.redpanda, True, local_retention, record_size, msg_count
         )
 
         # Confirm messages written
@@ -655,11 +652,10 @@ def test_timequery_empty_local_log(self):
 
         total_segments = 3
         record_size = 1024
-        base_ts = 1664453149000
         msg_count = (self.log_segment_size * total_segments) // record_size
         local_retention = 1  # Any value works for this test.
         topic, timestamps = self._create_and_produce(
-            self.redpanda, True, local_retention, base_ts, record_size, msg_count
+            self.redpanda, True, local_retention, record_size, msg_count
         )
 
         # Confirm messages written
@@ -770,6 +766,7 @@ class TestReadReplicaTimeQuery(RedpandaTest):
 
     log_segment_size = 1024 * 1024
     topic_name = "panda-topic"
+    base_ts = int(time.time() - 600) * 1000
 
     def __init__(self, test_context):
         super(TestReadReplicaTimeQuery, self).__init__(
@@ -812,7 +809,7 @@ def create_read_replica_topic(self) -> None:
             return False
         return True
 
-    def setup_clusters(self, base_ts, num_messages=0, partition_count=1) -> None:
+    def setup_clusters(self, num_messages=0, partition_count=1) -> None:
         spec = TopicSpec(
             name=self.topic_name,
             partition_count=partition_count,
@@ -829,7 +826,7 @@ def setup_clusters(self, base_ts, num_messages=0, partition_count=1) -> None:
             msg_size=1024,
             msg_count=num_messages,
             batch_max_bytes=10240,
-            fake_timestamp_ms=base_ts,
+            fake_timestamp_ms=self.base_ts,
         )
 
         producer.start()
@@ -868,9 +865,8 @@ def query_timestamp(self, ts, kcat_src, kcat_rr):
 
     @ducktape_cluster(num_nodes=7)
     def test_timequery(self):
-        base_ts = 1664453149000
         num_messages = 1000
-        self.setup_clusters(base_ts, num_messages, 3)
+        self.setup_clusters(num_messages, 3)
 
         def read_replica_ready():
             orig = RpkTool(self.redpanda).describe_topic(self.topic_name)
@@ -889,4 +885,4 @@ def read_replica_ready():
         kcat1 = KafkaCat(self.redpanda)
         kcat2 = KafkaCat(self.rr_cluster)
         for ix in range(0, num_messages, 20):
-            self.query_timestamp(base_ts + ix, kcat1, kcat2)
+            self.query_timestamp(self.base_ts + ix, kcat1, kcat2)