MINOR: Replication system tests should cover compressed path
Author: Jason Gustafson
Reviewers: Apurva Mehta Ismael Juma
Closes from hachikuji/add-replication-testcase-for-compression
@@ -138,10 +138,11 @@ class ReplicationTest(ProduceConsumeValidateTest):
@parametrize(failure_mode="hard_bounce",
broker_type="leader",
security_protocol="SASL_SSL", client_sasl_mechanism="SCRAM-SHA-256", interbroker_sasl_mechanism="SCRAM-SHA-512")
- def test_replication_with_broker_failure(self, failure_mode, security_protocol,
- broker_type, client_sasl_mechanism="GSSAPI",
- interbroker_sasl_mechanism="GSSAPI",
- enable_idempotence=False):
+ @matrix(failure_mode=["clean_shutdown", "hard_shutdown", "clean_bounce", "hard_bounce"],
+ security_protocol=["PLAINTEXT"], broker_type=["leader"], compression_type=["gzip"])
+ def test_replication_with_broker_failure(self, failure_mode, security_protocol, broker_type,
+ client_sasl_mechanism="GSSAPI", interbroker_sasl_mechanism="GSSAPI",
+ compression_type=None, enable_idempotence=False):
"""Replication tests.
These tests verify that replication provides simple durability guarantees by checking that data acked by
brokers is still available for consumption in the face of various failure scenarios.
@@ -161,7 +162,10 @@ class ReplicationTest(ProduceConsumeValidateTest):
self.kafka.interbroker_sasl_mechanism = interbroker_sasl_mechanism
new_consumer = False if self.kafka.security_protocol == "PLAINTEXT" else True
self.enable_idempotence = enable_idempotence
- self.producer = VerifiableProducer(self.test_context, self.num_producers, self.kafka, self.topic, throughput=self.producer_throughput, enable_idempotence=enable_idempotence)
+ compression_types = None if not compression_type else [compression_type] * self.num_producers
+ self.producer = VerifiableProducer(self.test_context, self.num_producers, self.kafka, self.topic,
+ throughput=self.producer_throughput, compression_types=compression_types,
+ enable_idempotence=enable_idempotence)
self.consumer = ConsoleConsumer(self.test_context, self.num_consumers, self.kafka, self.topic, new_consumer=new_consumer, consumer_timeout_ms=60000, message_validator=is_int)
self.kafka.start()
self.run_produce_consume_validate(core_test_action=lambda: failures[failure_mode](self, broker_type))
