From ec383d8f675c60b23ff706055d76175f3fa5afb3 Mon Sep 17 00:00:00 2001 From: Ron Dagostino Date: Thu, 25 Feb 2021 13:54:57 -0500 Subject: [PATCH 1/2] MINOR: fix syntax error in upgrade_test.py --- tests/kafkatest/tests/core/upgrade_test.py | 87 +++++++++++----------- 1 file changed, 44 insertions(+), 43 deletions(-) diff --git a/tests/kafkatest/tests/core/upgrade_test.py b/tests/kafkatest/tests/core/upgrade_test.py index 183e4900e8c1d..a1cbcfbf19dc0 100644 --- a/tests/kafkatest/tests/core/upgrade_test.py +++ b/tests/kafkatest/tests/core/upgrade_test.py @@ -77,47 +77,47 @@ def perform_upgrade(self, from_kafka_version, to_message_format_version=None): @cluster(num_nodes=6) @parametrize(from_kafka_version=str(LATEST_2_7), to_message_format_version=None, compression_types=["none"]) - @parametrize(from_kafka_version=str(LATEST_2_7), to_message_format_version=None, compression_types=["lz4"]) - @parametrize(from_kafka_version=str(LATEST_2_7), to_message_format_version=None, compression_types=["snappy"]) - @parametrize(from_kafka_version=str(LATEST_2_6), to_message_format_version=None, compression_types=["none"]) - @parametrize(from_kafka_version=str(LATEST_2_6), to_message_format_version=None, compression_types=["lz4"]) - @parametrize(from_kafka_version=str(LATEST_2_6), to_message_format_version=None, compression_types=["snappy"]) - @parametrize(from_kafka_version=str(LATEST_2_5), to_message_format_version=None, compression_types=["none"]) - @parametrize(from_kafka_version=str(LATEST_2_5), to_message_format_version=None, compression_types=["zstd"]) - @parametrize(from_kafka_version=str(LATEST_2_4), to_message_format_version=None, compression_types=["none"]) - @parametrize(from_kafka_version=str(LATEST_2_4), to_message_format_version=None, compression_types=["zstd"]) - @parametrize(from_kafka_version=str(LATEST_2_3), to_message_format_version=None, compression_types=["none"]) - @parametrize(from_kafka_version=str(LATEST_2_3), to_message_format_version=None, compression_types=["zstd"]) - @parametrize(from_kafka_version=str(LATEST_2_2), to_message_format_version=None, compression_types=["none"]) - @parametrize(from_kafka_version=str(LATEST_2_2), to_message_format_version=None, compression_types=["zstd"]) - @parametrize(from_kafka_version=str(LATEST_2_1), to_message_format_version=None, compression_types=["none"]) - @parametrize(from_kafka_version=str(LATEST_2_1), to_message_format_version=None, compression_types=["lz4"]) - @parametrize(from_kafka_version=str(LATEST_2_0), to_message_format_version=None, compression_types=["none"]) - @parametrize(from_kafka_version=str(LATEST_2_0), to_message_format_version=None, compression_types=["snappy"]) - @parametrize(from_kafka_version=str(LATEST_1_1), to_message_format_version=None, compression_types=["none"]) - @parametrize(from_kafka_version=str(LATEST_1_1), to_message_format_version=None, compression_types=["lz4"]) - @parametrize(from_kafka_version=str(LATEST_1_0), to_message_format_version=None, compression_types=["none"]) - @parametrize(from_kafka_version=str(LATEST_1_0), to_message_format_version=None, compression_types=["snappy"]) - @parametrize(from_kafka_version=str(LATEST_0_11_0), to_message_format_version=None, compression_types=["gzip"]) - @parametrize(from_kafka_version=str(LATEST_0_11_0), to_message_format_version=None, compression_types=["lz4"]) - @parametrize(from_kafka_version=str(LATEST_0_10_2), to_message_format_version=str(LATEST_0_9), compression_types=["none"]) - @parametrize(from_kafka_version=str(LATEST_0_10_2), to_message_format_version=str(LATEST_0_10), compression_types=["snappy"]) - @parametrize(from_kafka_version=str(LATEST_0_10_2), to_message_format_version=None, compression_types=["none"]) - @parametrize(from_kafka_version=str(LATEST_0_10_2), to_message_format_version=None, compression_types=["lz4"]) - @parametrize(from_kafka_version=str(LATEST_0_10_1), to_message_format_version=None, compression_types=["lz4"]) - @parametrize(from_kafka_version=str(LATEST_0_10_1), to_message_format_version=None, compression_types=["snappy"]) - @parametrize(from_kafka_version=str(LATEST_0_10_0), to_message_format_version=None, compression_types=["snappy"]) - @parametrize(from_kafka_version=str(LATEST_0_10_0), to_message_format_version=None, compression_types=["lz4"]) - @cluster(num_nodes=7) - @parametrize(from_kafka_version=str(LATEST_0_9), to_message_format_version=None, compression_types=["none"], security_protocol="SASL_SSL") - @cluster(num_nodes=6) - @parametrize(from_kafka_version=str(LATEST_0_9), to_message_format_version=None, compression_types=["snappy"]) - @parametrize(from_kafka_version=str(LATEST_0_9), to_message_format_version=None, compression_types=["lz4"]) - @parametrize(from_kafka_version=str(LATEST_0_9), to_message_format_version=str(LATEST_0_9), compression_types=["none"]) - @parametrize(from_kafka_version=str(LATEST_0_9), to_message_format_version=str(LATEST_0_9), compression_types=["lz4"]) - @cluster(num_nodes=7) - @parametrize(from_kafka_version=str(LATEST_0_8_2), to_message_format_version=None, compression_types=["none"]) - @parametrize(from_kafka_version=str(LATEST_0_8_2), to_message_format_version=None, compression_types=["snappy"]) + # @parametrize(from_kafka_version=str(LATEST_2_7), to_message_format_version=None, compression_types=["lz4"]) + # @parametrize(from_kafka_version=str(LATEST_2_7), to_message_format_version=None, compression_types=["snappy"]) + # @parametrize(from_kafka_version=str(LATEST_2_6), to_message_format_version=None, compression_types=["none"]) + # @parametrize(from_kafka_version=str(LATEST_2_6), to_message_format_version=None, compression_types=["lz4"]) + # @parametrize(from_kafka_version=str(LATEST_2_6), to_message_format_version=None, compression_types=["snappy"]) + # @parametrize(from_kafka_version=str(LATEST_2_5), to_message_format_version=None, compression_types=["none"]) + # @parametrize(from_kafka_version=str(LATEST_2_5), to_message_format_version=None, compression_types=["zstd"]) + # @parametrize(from_kafka_version=str(LATEST_2_4), to_message_format_version=None, compression_types=["none"]) + # @parametrize(from_kafka_version=str(LATEST_2_4), to_message_format_version=None, compression_types=["zstd"]) + # @parametrize(from_kafka_version=str(LATEST_2_3), to_message_format_version=None, compression_types=["none"]) + # @parametrize(from_kafka_version=str(LATEST_2_3), to_message_format_version=None, compression_types=["zstd"]) + # @parametrize(from_kafka_version=str(LATEST_2_2), to_message_format_version=None, compression_types=["none"]) + # @parametrize(from_kafka_version=str(LATEST_2_2), to_message_format_version=None, compression_types=["zstd"]) + # @parametrize(from_kafka_version=str(LATEST_2_1), to_message_format_version=None, compression_types=["none"]) + # @parametrize(from_kafka_version=str(LATEST_2_1), to_message_format_version=None, compression_types=["lz4"]) + # @parametrize(from_kafka_version=str(LATEST_2_0), to_message_format_version=None, compression_types=["none"]) + # @parametrize(from_kafka_version=str(LATEST_2_0), to_message_format_version=None, compression_types=["snappy"]) + # @parametrize(from_kafka_version=str(LATEST_1_1), to_message_format_version=None, compression_types=["none"]) + # @parametrize(from_kafka_version=str(LATEST_1_1), to_message_format_version=None, compression_types=["lz4"]) + # @parametrize(from_kafka_version=str(LATEST_1_0), to_message_format_version=None, compression_types=["none"]) + # @parametrize(from_kafka_version=str(LATEST_1_0), to_message_format_version=None, compression_types=["snappy"]) + # @parametrize(from_kafka_version=str(LATEST_0_11_0), to_message_format_version=None, compression_types=["gzip"]) + # @parametrize(from_kafka_version=str(LATEST_0_11_0), to_message_format_version=None, compression_types=["lz4"]) + # @parametrize(from_kafka_version=str(LATEST_0_10_2), to_message_format_version=str(LATEST_0_9), compression_types=["none"]) + # @parametrize(from_kafka_version=str(LATEST_0_10_2), to_message_format_version=str(LATEST_0_10), compression_types=["snappy"]) + # @parametrize(from_kafka_version=str(LATEST_0_10_2), to_message_format_version=None, compression_types=["none"]) + # @parametrize(from_kafka_version=str(LATEST_0_10_2), to_message_format_version=None, compression_types=["lz4"]) + # @parametrize(from_kafka_version=str(LATEST_0_10_1), to_message_format_version=None, compression_types=["lz4"]) + # @parametrize(from_kafka_version=str(LATEST_0_10_1), to_message_format_version=None, compression_types=["snappy"]) + # @parametrize(from_kafka_version=str(LATEST_0_10_0), to_message_format_version=None, compression_types=["snappy"]) + # @parametrize(from_kafka_version=str(LATEST_0_10_0), to_message_format_version=None, compression_types=["lz4"]) + # @cluster(num_nodes=7) + # @parametrize(from_kafka_version=str(LATEST_0_9), to_message_format_version=None, compression_types=["none"], security_protocol="SASL_SSL") + # @cluster(num_nodes=6) + # @parametrize(from_kafka_version=str(LATEST_0_9), to_message_format_version=None, compression_types=["snappy"]) + # @parametrize(from_kafka_version=str(LATEST_0_9), to_message_format_version=None, compression_types=["lz4"]) + # @parametrize(from_kafka_version=str(LATEST_0_9), to_message_format_version=str(LATEST_0_9), compression_types=["none"]) + # @parametrize(from_kafka_version=str(LATEST_0_9), to_message_format_version=str(LATEST_0_9), compression_types=["lz4"]) + # @cluster(num_nodes=7) + # @parametrize(from_kafka_version=str(LATEST_0_8_2), to_message_format_version=None, compression_types=["none"]) + # @parametrize(from_kafka_version=str(LATEST_0_8_2), to_message_format_version=None, compression_types=["snappy"]) def test_upgrade(self, from_kafka_version, to_message_format_version, compression_types, security_protocol="PLAINTEXT"): """Test upgrade of Kafka broker cluster from various versions to the current version @@ -139,8 +139,9 @@ def test_upgrade(self, from_kafka_version, to_message_format_version, compressio - Finally, validate that every message acked by the producer was consumed by the consumer """ self.zk = ZookeeperService(self.test_context, num_nodes=1, version=KafkaVersion(from_kafka_version)) + fromKafkaVersion = KafkaVersion(from_kafka_version) self.kafka = KafkaService(self.test_context, num_nodes=3, zk=self.zk, - version=KafkaVersion(from_kafka_version), + version=fromKafkaVersion, topics={self.topic: {"partitions": self.partitions, "replication-factor": self.replication_factor, 'configs': {"min.insync.replicas": 2}}}) @@ -171,7 +172,7 @@ def test_upgrade(self, from_kafka_version, to_message_format_version, compressio # after leader change. Tolerate limited data loss for this case to avoid transient test failures. self.may_truncate_acked_records = False if from_kafka_version >= V_0_11_0_0 else True - new_consumer = from_kafka_version.consumer_supports_bootstrap_server() + new_consumer = fromKafkaVersion.consumer_supports_bootstrap_server() # TODO - reduce the timeout self.consumer = ConsoleConsumer(self.test_context, self.num_consumers, self.kafka, self.topic, new_consumer=new_consumer, consumer_timeout_ms=30000, From 47f489892376f005ffdbcfbecc3fd8fb5b8a69ad Mon Sep 17 00:00:00 2001 From: Ron Dagostino Date: Thu, 25 Feb 2021 14:01:39 -0500 Subject: [PATCH 2/2] fixup --- tests/kafkatest/tests/core/upgrade_test.py | 82 +++++++++++----------- 1 file changed, 41 insertions(+), 41 deletions(-) diff --git a/tests/kafkatest/tests/core/upgrade_test.py b/tests/kafkatest/tests/core/upgrade_test.py index a1cbcfbf19dc0..59db68f692b73 100644 --- a/tests/kafkatest/tests/core/upgrade_test.py +++ b/tests/kafkatest/tests/core/upgrade_test.py @@ -77,47 +77,47 @@ def perform_upgrade(self, from_kafka_version, to_message_format_version=None): @cluster(num_nodes=6) @parametrize(from_kafka_version=str(LATEST_2_7), to_message_format_version=None, compression_types=["none"]) - # @parametrize(from_kafka_version=str(LATEST_2_7), to_message_format_version=None, compression_types=["lz4"]) - # @parametrize(from_kafka_version=str(LATEST_2_7), to_message_format_version=None, compression_types=["snappy"]) - # @parametrize(from_kafka_version=str(LATEST_2_6), to_message_format_version=None, compression_types=["none"]) - # @parametrize(from_kafka_version=str(LATEST_2_6), to_message_format_version=None, compression_types=["lz4"]) - # @parametrize(from_kafka_version=str(LATEST_2_6), to_message_format_version=None, compression_types=["snappy"]) - # @parametrize(from_kafka_version=str(LATEST_2_5), to_message_format_version=None, compression_types=["none"]) - # @parametrize(from_kafka_version=str(LATEST_2_5), to_message_format_version=None, compression_types=["zstd"]) - # @parametrize(from_kafka_version=str(LATEST_2_4), to_message_format_version=None, compression_types=["none"]) - # @parametrize(from_kafka_version=str(LATEST_2_4), to_message_format_version=None, compression_types=["zstd"]) - # @parametrize(from_kafka_version=str(LATEST_2_3), to_message_format_version=None, compression_types=["none"]) - # @parametrize(from_kafka_version=str(LATEST_2_3), to_message_format_version=None, compression_types=["zstd"]) - # @parametrize(from_kafka_version=str(LATEST_2_2), to_message_format_version=None, compression_types=["none"]) - # @parametrize(from_kafka_version=str(LATEST_2_2), to_message_format_version=None, compression_types=["zstd"]) - # @parametrize(from_kafka_version=str(LATEST_2_1), to_message_format_version=None, compression_types=["none"]) - # @parametrize(from_kafka_version=str(LATEST_2_1), to_message_format_version=None, compression_types=["lz4"]) - # @parametrize(from_kafka_version=str(LATEST_2_0), to_message_format_version=None, compression_types=["none"]) - # @parametrize(from_kafka_version=str(LATEST_2_0), to_message_format_version=None, compression_types=["snappy"]) - # @parametrize(from_kafka_version=str(LATEST_1_1), to_message_format_version=None, compression_types=["none"]) - # @parametrize(from_kafka_version=str(LATEST_1_1), to_message_format_version=None, compression_types=["lz4"]) - # @parametrize(from_kafka_version=str(LATEST_1_0), to_message_format_version=None, compression_types=["none"]) - # @parametrize(from_kafka_version=str(LATEST_1_0), to_message_format_version=None, compression_types=["snappy"]) - # @parametrize(from_kafka_version=str(LATEST_0_11_0), to_message_format_version=None, compression_types=["gzip"]) - # @parametrize(from_kafka_version=str(LATEST_0_11_0), to_message_format_version=None, compression_types=["lz4"]) - # @parametrize(from_kafka_version=str(LATEST_0_10_2), to_message_format_version=str(LATEST_0_9), compression_types=["none"]) - # @parametrize(from_kafka_version=str(LATEST_0_10_2), to_message_format_version=str(LATEST_0_10), compression_types=["snappy"]) - # @parametrize(from_kafka_version=str(LATEST_0_10_2), to_message_format_version=None, compression_types=["none"]) - # @parametrize(from_kafka_version=str(LATEST_0_10_2), to_message_format_version=None, compression_types=["lz4"]) - # @parametrize(from_kafka_version=str(LATEST_0_10_1), to_message_format_version=None, compression_types=["lz4"]) - # @parametrize(from_kafka_version=str(LATEST_0_10_1), to_message_format_version=None, compression_types=["snappy"]) - # @parametrize(from_kafka_version=str(LATEST_0_10_0), to_message_format_version=None, compression_types=["snappy"]) - # @parametrize(from_kafka_version=str(LATEST_0_10_0), to_message_format_version=None, compression_types=["lz4"]) - # @cluster(num_nodes=7) - # @parametrize(from_kafka_version=str(LATEST_0_9), to_message_format_version=None, compression_types=["none"], security_protocol="SASL_SSL") - # @cluster(num_nodes=6) - # @parametrize(from_kafka_version=str(LATEST_0_9), to_message_format_version=None, compression_types=["snappy"]) - # @parametrize(from_kafka_version=str(LATEST_0_9), to_message_format_version=None, compression_types=["lz4"]) - # @parametrize(from_kafka_version=str(LATEST_0_9), to_message_format_version=str(LATEST_0_9), compression_types=["none"]) - # @parametrize(from_kafka_version=str(LATEST_0_9), to_message_format_version=str(LATEST_0_9), compression_types=["lz4"]) - # @cluster(num_nodes=7) - # @parametrize(from_kafka_version=str(LATEST_0_8_2), to_message_format_version=None, compression_types=["none"]) - # @parametrize(from_kafka_version=str(LATEST_0_8_2), to_message_format_version=None, compression_types=["snappy"]) + @parametrize(from_kafka_version=str(LATEST_2_7), to_message_format_version=None, compression_types=["lz4"]) + @parametrize(from_kafka_version=str(LATEST_2_7), to_message_format_version=None, compression_types=["snappy"]) + @parametrize(from_kafka_version=str(LATEST_2_6), to_message_format_version=None, compression_types=["none"]) + @parametrize(from_kafka_version=str(LATEST_2_6), to_message_format_version=None, compression_types=["lz4"]) + @parametrize(from_kafka_version=str(LATEST_2_6), to_message_format_version=None, compression_types=["snappy"]) + @parametrize(from_kafka_version=str(LATEST_2_5), to_message_format_version=None, compression_types=["none"]) + @parametrize(from_kafka_version=str(LATEST_2_5), to_message_format_version=None, compression_types=["zstd"]) + @parametrize(from_kafka_version=str(LATEST_2_4), to_message_format_version=None, compression_types=["none"]) + @parametrize(from_kafka_version=str(LATEST_2_4), to_message_format_version=None, compression_types=["zstd"]) + @parametrize(from_kafka_version=str(LATEST_2_3), to_message_format_version=None, compression_types=["none"]) + @parametrize(from_kafka_version=str(LATEST_2_3), to_message_format_version=None, compression_types=["zstd"]) + @parametrize(from_kafka_version=str(LATEST_2_2), to_message_format_version=None, compression_types=["none"]) + @parametrize(from_kafka_version=str(LATEST_2_2), to_message_format_version=None, compression_types=["zstd"]) + @parametrize(from_kafka_version=str(LATEST_2_1), to_message_format_version=None, compression_types=["none"]) + @parametrize(from_kafka_version=str(LATEST_2_1), to_message_format_version=None, compression_types=["lz4"]) + @parametrize(from_kafka_version=str(LATEST_2_0), to_message_format_version=None, compression_types=["none"]) + @parametrize(from_kafka_version=str(LATEST_2_0), to_message_format_version=None, compression_types=["snappy"]) + @parametrize(from_kafka_version=str(LATEST_1_1), to_message_format_version=None, compression_types=["none"]) + @parametrize(from_kafka_version=str(LATEST_1_1), to_message_format_version=None, compression_types=["lz4"]) + @parametrize(from_kafka_version=str(LATEST_1_0), to_message_format_version=None, compression_types=["none"]) + @parametrize(from_kafka_version=str(LATEST_1_0), to_message_format_version=None, compression_types=["snappy"]) + @parametrize(from_kafka_version=str(LATEST_0_11_0), to_message_format_version=None, compression_types=["gzip"]) + @parametrize(from_kafka_version=str(LATEST_0_11_0), to_message_format_version=None, compression_types=["lz4"]) + @parametrize(from_kafka_version=str(LATEST_0_10_2), to_message_format_version=str(LATEST_0_9), compression_types=["none"]) + @parametrize(from_kafka_version=str(LATEST_0_10_2), to_message_format_version=str(LATEST_0_10), compression_types=["snappy"]) + @parametrize(from_kafka_version=str(LATEST_0_10_2), to_message_format_version=None, compression_types=["none"]) + @parametrize(from_kafka_version=str(LATEST_0_10_2), to_message_format_version=None, compression_types=["lz4"]) + @parametrize(from_kafka_version=str(LATEST_0_10_1), to_message_format_version=None, compression_types=["lz4"]) + @parametrize(from_kafka_version=str(LATEST_0_10_1), to_message_format_version=None, compression_types=["snappy"]) + @parametrize(from_kafka_version=str(LATEST_0_10_0), to_message_format_version=None, compression_types=["snappy"]) + @parametrize(from_kafka_version=str(LATEST_0_10_0), to_message_format_version=None, compression_types=["lz4"]) + @cluster(num_nodes=7) + @parametrize(from_kafka_version=str(LATEST_0_9), to_message_format_version=None, compression_types=["none"], security_protocol="SASL_SSL") + @cluster(num_nodes=6) + @parametrize(from_kafka_version=str(LATEST_0_9), to_message_format_version=None, compression_types=["snappy"]) + @parametrize(from_kafka_version=str(LATEST_0_9), to_message_format_version=None, compression_types=["lz4"]) + @parametrize(from_kafka_version=str(LATEST_0_9), to_message_format_version=str(LATEST_0_9), compression_types=["none"]) + @parametrize(from_kafka_version=str(LATEST_0_9), to_message_format_version=str(LATEST_0_9), compression_types=["lz4"]) + @cluster(num_nodes=7) + @parametrize(from_kafka_version=str(LATEST_0_8_2), to_message_format_version=None, compression_types=["none"]) + @parametrize(from_kafka_version=str(LATEST_0_8_2), to_message_format_version=None, compression_types=["snappy"]) def test_upgrade(self, from_kafka_version, to_message_format_version, compression_types, security_protocol="PLAINTEXT"): """Test upgrade of Kafka broker cluster from various versions to the current version