feat(tests): test s3 wal in automq enhanced e2e tests (AutoMQ#1712)

* feat(tests): enable s3 wal in e2e test

Signed-off-by: SSpirits <admin@lv5.moe>

* feat(tests): test s3 wal in automq enhanced e2e tests

Signed-off-by: SSpirits <admin@lv5.moe>

---------

Signed-off-by: SSpirits <admin@lv5.moe>

Author: ShadowySpirits

tests/kafkatest/automq/autobalancer_test.py

@@ -9,7 +9,8 @@
 from ducktape.mark import parametrize
 from ducktape.mark.resource import cluster
 from ducktape.tests.test import Test
-from kafkatest.automq.automq_e2e_util import (run_simple_load, TOPIC, append_info)
+from ducktape.mark import matrix
+from kafkatest.automq.automq_e2e_util import (FILE_WAL, S3_WAL, run_simple_load, TOPIC, append_info)
 from kafkatest.services.kafka import KafkaService
 
 # Configuration constants for the AutoBalancer
@@ -94,7 +95,7 @@ def __init__(self, test_context):
         self.avg_deviation = 0.2
         self.maximum_broker_deviation_percentage = 0.15
 
-    def create_kafka(self, num_nodes=1, partition=1, exclude_broker=None, exclude_topic=None, replica_assignment=None):
+    def create_kafka(self, num_nodes=1, partition=1, exclude_broker=None, exclude_topic=None, replica_assignment=None, wal='file'):
         """
         Create and configure a Kafka cluster for testing.
 
@@ -123,6 +124,7 @@ def create_kafka(self, num_nodes=1, partition=1, exclude_broker=None, exclude_to
             [REPORT_INTERVAL, str(4000)],
             [DETECT_INTERVAL, str(8000)],
             [METRIC_REPORTERS, 'kafka.autobalancer.metricsreporter.AutoBalancerMetricsReporter'],
+            ['s3.wal.path', FILE_WAL if wal == 'file' else S3_WAL],
         ]
 
         if exclude_broker:
@@ -153,16 +155,16 @@ def create_kafka(self, num_nodes=1, partition=1, exclude_broker=None, exclude_to
         self.start = True
 
     @cluster(num_nodes=5)
-    @parametrize(automq_num_nodes=2, partition=4, replica_assignment='1,1,1,2')
-    def test_action(self, automq_num_nodes, partition, replica_assignment):
+    @matrix(automq_num_nodes=[2], partition=[4], replica_assignment=['1,1,1,2'], wal=['file', 's3'])
+    def test_action(self, automq_num_nodes, partition, replica_assignment, wal):
         """
         Test throughput distribution across brokers
         :param automq_num_nodes: Number of automq
         :param partition: Number of partitions
         :param replica_assignment: Replica assignment for partitions
         """
         success, msg = True, ''
-        self.create_kafka(num_nodes=automq_num_nodes, partition=partition, replica_assignment=replica_assignment)
+        self.create_kafka(num_nodes=automq_num_nodes, partition=partition, replica_assignment=replica_assignment, wal=wal)
         self.kafka.start()
 
         run_simple_load(test_context=self.context, kafka=self.kafka, logger=self.logger, topic=self.topic,
@@ -176,8 +178,8 @@ def test_action(self, automq_num_nodes, partition, replica_assignment):
         assert success, msg
 
     @cluster(num_nodes=4)
-    @parametrize(automq_num_nodes=2, exclude_broker='2', partition=4, replica_assignment='1,1,1,2')
-    def test_broker_white_list(self, automq_num_nodes, exclude_broker, partition, replica_assignment):
+    @matrix(automq_num_nodes=[2], exclude_broker=['2'], partition=[4], replica_assignment=['1,1,1,2'], wal=['file', 's3'])
+    def test_broker_white_list(self, automq_num_nodes, exclude_broker, partition, replica_assignment, wal):
         """
         Test broker exclusion functionality
         :param automq_num_nodes: Number of automq
@@ -187,22 +189,23 @@ def test_broker_white_list(self, automq_num_nodes, exclude_broker, partition, re
         """
         success, msg = True, ''
         self.create_kafka(num_nodes=automq_num_nodes, exclude_broker=exclude_broker, partition=partition,
-                          replica_assignment=replica_assignment)
+                          replica_assignment=replica_assignment, wal=wal)
         self.kafka.start()
         before = self.kafka.parse_describe_topic(self.kafka.describe_topic(TOPIC))
         run_simple_load(test_context=self.context, kafka=self.kafka, logger=self.logger, topic=self.topic,
                         num_records=20000, throughput=1300)
         after = self.kafka.parse_describe_topic(self.kafka.describe_topic(TOPIC))
 
-        success_, msg_ = check_partition_eq(topic_info1=before, topic_info2=after)
+        success_, msg_ = check_partition_eq(topic_info1=before,
+         topic_info2=after)
         success = success and success_
         msg = append_info(msg, success_, msg_)
 
         assert success, msg
 
     @cluster(num_nodes=6)
-    @parametrize(automq_num_nodes=2)
-    def test_topic_white_list(self, automq_num_nodes):
+    @matrix(automq_num_nodes=[2], wal=['file', 's3'])
+    def test_topic_white_list(self, automq_num_nodes, wal):
         """
         Test topic exclusion functionality
         :param automq_num_nodes: Number of automq
@@ -224,7 +227,7 @@ def test_topic_white_list(self, automq_num_nodes):
             "configs": {"min.insync.replicas": 1},
             "replica-assignment": '1,1,1,2',
         }
-        self.create_kafka(num_nodes=automq_num_nodes, exclude_topic=topic1, partition=1, replica_assignment='1')
+        self.create_kafka(num_nodes=automq_num_nodes, exclude_topic=topic1, partition=1, replica_assignment='1', wal=wal)
         self.kafka.start()
         self.kafka.create_topic(topic_cfg1)
         self.kafka.create_topic(topic_cfg2)

tests/kafkatest/automq/automq_e2e_util.py

@@ -123,6 +123,8 @@ def publish_broker_configuration(kafka, producer_byte_rate, consumer_byte_rate,
 JMX_TOPIC_IN = f'kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec,topic={TOPIC}'
 JMX_TOPIC_OUT = f'kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec,topic={TOPIC}'
 JMX_ONE_MIN = ':OneMinuteRate'
+FILE_WAL = '0@file:///mnt/kafka/kafka-data-logs-1/s3wal'
+S3_WAL = '0@s3://ko3?region=us-east-1&endpoint=http://10.5.0.2:4566&pathStyle=false&batchInterval=100&maxBytesInBatch=4194304&maxUnflushedBytes=1073741824&maxInflightUploadCount=50'
 
 
 def run_perf_producer(test_context, kafka, num_records=RECORD_NUM, throughput=DEFAULT_THROUGHPUT,

tests/kafkatest/automq/memory_occupancy_test.py

@@ -19,7 +19,7 @@
 from kafkatest.services.kafka import KafkaService
 from kafkatest.version import DEV_BRANCH, KafkaVersion
 from kafkatest.services.performance import ProducerPerformanceService, ConsumerPerformanceService
-from kafkatest.automq.automq_e2e_util import formatted_time, parse_log_entry, parse_producer_performance_stdout
+from kafkatest.automq.automq_e2e_util import FILE_WAL, S3_WAL, formatted_time, parse_log_entry, parse_producer_performance_stdout
 
 
 class TestJVMMemoryOccupancy(Test):
@@ -35,7 +35,7 @@ def __init__(self, test_context):
         self.consume_group = 'test_group'
         self.records_consumed = []
 
-    def create_kafka(self, num_nodes=1, partition=None, log_size=None, block_size=None, **kwargs):
+    def create_kafka(self, num_nodes=1, partition=None, log_size=None, block_size=None, wal='file', **kwargs):
         """
         Create and configure Kafka service.
 
@@ -52,7 +52,8 @@ def create_kafka(self, num_nodes=1, partition=None, log_size=None, block_size=No
             ['s3.wal.cache.size', str(log_size)],
             ['s3.wal.capacity', str(log_size)],
             ['s3.wal.upload.threshold', str(log_size // 4)],
-            ['s3.block.cache.size', str(block_size)]
+            ['s3.block.cache.size', str(block_size)],
+            ['s3.wal.path', FILE_WAL if wal == 'file' else S3_WAL],
         ]
 
         self.kafka = KafkaService(
@@ -105,8 +106,8 @@ def check_the_consumption_quantity(self, records):
         assert int(receive_num) == records, f"Receive count does not match the expected records count: expected {records}, but got {receive_num}"
 
     @cluster(num_nodes=3)
-    @matrix(partition=[128, 512], log_size=[256 * 1024 * 1024], block_size=[128 * 1024 * 1024, 256 * 1024 * 1024])
-    def test(self, partition, log_size, block_size):
+    @matrix(partition=[128, 512], log_size=[256 * 1024 * 1024], block_size=[128 * 1024 * 1024, 256 * 1024 * 1024], wal=['file', 's3'])
+    def test(self, partition, log_size, block_size, wal):
         """
         At any time, 1/writable record in Metric<=log cache size+100MB
         At any time, 11/block_cache in Metric<=block cache size
@@ -116,7 +117,7 @@ def test(self, partition, log_size, block_size):
         :param block_size: Block size for Kafka configuration.
         """
         # Start Kafka
-        self.create_kafka(partition=partition, log_size=log_size, block_size=block_size)
+        self.create_kafka(partition=partition, log_size=log_size, block_size=block_size, wal=wal)
         self.kafka.start()
 
         # Check Kafka configuration

tests/kafkatest/automq/quota_test.py

@@ -20,7 +20,7 @@
 from kafkatest.services.kafka import KafkaService
 from kafkatest.services.performance import ProducerPerformanceService
 from kafkatest.version import DEV_BRANCH
-from kafkatest.automq.automq_e2e_util import publish_broker_configuration
+from kafkatest.automq.automq_e2e_util import FILE_WAL, S3_WAL, publish_broker_configuration
 
 
 class QuotaTest(Test):
@@ -46,7 +46,7 @@ def __init__(self, test_context):
         self.success = True
         self.msg = ''
 
-    def create_kafka(self, test_context, broker_quota_in, broker_quota_out):
+    def create_kafka(self, test_context, broker_quota_in, broker_quota_out, broker_wal):
         log_size = 256 * 1024 * 1024
         block_size = 256 * 1024 * 1024
         server_prop_overrides = [
@@ -57,6 +57,7 @@ def create_kafka(self, test_context, broker_quota_in, broker_quota_out):
             ['s3.wal.capacity', str(log_size)],
             ['s3.wal.upload.threshold', str(log_size // 4)],
             ['s3.block.cache.size', str(block_size)],
+            ['s3.wal.path', FILE_WAL if broker_wal == 'file' else S3_WAL],
         ]
         self.kafka = KafkaService(test_context, num_nodes=1, zk=None,
                                   kafka_heap_opts="-Xmx2048m -Xms2048m",
@@ -100,9 +101,9 @@ def start_console_consumer(self):
             assert len(messages) > 0, "consumer %d didn't consume any message before timeout" % idx
 
     @cluster(num_nodes=5)
-    @matrix(broker_in=[2500000], broker_out=[2000000])
-    def test_quota(self, broker_in, broker_out):
-        self.create_kafka(self.test_context, broker_in, broker_out)
+    @matrix(broker_in=[2500000], broker_out=[2000000], wal=['file', 's3'])
+    def test_quota(self, broker_in, broker_out, wal):
+        self.create_kafka(self.test_context, broker_in, broker_out, wal)
         self.kafka.start()
         records = 50000
         self.logger.info(f'update to {broker_in},{broker_out}')