root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Revert "KAFKA-16275: Update kraft_upgrade_test.py to support KIP-848’s group protocol config (#16409) (#16441) 

This reverts commit e95e91a.

With the change to include the group.version flag, these tests fail due to trying to set the feature for the old version.

It is unclear if these tests originally worked as intended and given the upgrade is not expected for 3.8, we will just revert from 3.8.

Reviewers: David Jacot <djacot@confluent.io>
This commit is contained in:
Justine Olshan 2024-06-25 10:15:58 -07:00 committed by GitHub
parent ceec218351
commit a599b89fe0
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 11 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
tests/kafkatest/tests/core/kraft_upgrade_test.py
View File

@ -17,13 +17,13 @@ from ducktape.mark import parametrize, matrix
from ducktape.mark.resource import cluster
from ducktape.utils.util import wait_until
from kafkatest.services.console_consumer import ConsoleConsumer
from kafkatest.services.kafka import KafkaService, consumer_group
from kafkatest.services.kafka import KafkaService
from kafkatest.services.kafka.quorum import isolated_kraft, combined_kraft
from kafkatest.services.verifiable_producer import VerifiableProducer
from kafkatest.tests.produce_consume_validate import ProduceConsumeValidateTest
from kafkatest.utils import is_int
from kafkatest.version import LATEST_3_1, LATEST_3_2, LATEST_3_3, LATEST_3_4, LATEST_3_5, LATEST_3_7, \
DEV_BRANCH, KafkaVersion, LATEST_STABLE_METADATA_VERSION
from kafkatest.version import LATEST_3_1, LATEST_3_2, LATEST_3_3, LATEST_3_4, LATEST_3_5, \
LATEST_3_6, LATEST_3_7, DEV_BRANCH, KafkaVersion, LATEST_STABLE_METADATA_VERSION
#
# Test upgrading between different KRaft versions.
@ -74,7 +74,7 @@ class TestKRaftUpgrade(ProduceConsumeValidateTest):
self.logger.info("Changing metadata.version to %s" % LATEST_STABLE_METADATA_VERSION)
self.kafka.upgrade_metadata_version(LATEST_STABLE_METADATA_VERSION)
def run_upgrade(self, from_kafka_version, group_protocol):
def run_upgrade(self, from_kafka_version):
"""Test upgrade of Kafka broker cluster from various versions to the current version
from_kafka_version is a Kafka version to upgrade from.
@ -101,8 +101,7 @@ class TestKRaftUpgrade(ProduceConsumeValidateTest):
version=KafkaVersion(from_kafka_version))
self.consumer = ConsoleConsumer(self.test_context, self.num_consumers, self.kafka,
self.topic, new_consumer=True, consumer_timeout_ms=30000,
message_validator=is_int, version=KafkaVersion(from_kafka_version),
consumer_properties=consumer_group.maybe_set_group_protocol(group_protocol))
message_validator=is_int, version=KafkaVersion(from_kafka_version))
self.run_produce_consume_validate(core_test_action=lambda: self.perform_version_change(from_kafka_version))
cluster_id = self.kafka.cluster_id()
assert cluster_id is not None
@ -110,24 +109,14 @@ class TestKRaftUpgrade(ProduceConsumeValidateTest):
assert self.kafka.check_protocol_errors(self)
@cluster(num_nodes=5)
@matrix(from_kafka_version=[str(LATEST_3_1), str(LATEST_3_2), str(LATEST_3_3), str(LATEST_3_4), str(LATEST_3_5), str(DEV_BRANCH)],
use_new_coordinator=[True, False],
@matrix(from_kafka_version=[str(LATEST_3_1), str(LATEST_3_2), str(LATEST_3_3), str(LATEST_3_4), str(LATEST_3_5), str(LATEST_3_6), str(LATEST_3_7), str(DEV_BRANCH)],
metadata_quorum=[combined_kraft])
@matrix(from_kafka_version=[str(LATEST_3_7), str(DEV_BRANCH)],
use_new_coordinator=[True],
metadata_quorum=[combined_kraft],
group_protocol=consumer_group.all_group_protocols)
def test_combined_mode_upgrade(self, from_kafka_version, metadata_quorum, use_new_coordinator=False, group_protocol=None):
self.run_upgrade(from_kafka_version, group_protocol)
def test_combined_mode_upgrade(self, from_kafka_version, metadata_quorum, use_new_coordinator=False):
self.run_upgrade(from_kafka_version)
@cluster(num_nodes=8)
@matrix(from_kafka_version=[str(LATEST_3_1), str(LATEST_3_2), str(LATEST_3_3), str(LATEST_3_4), str(LATEST_3_5), str(DEV_BRANCH)],
use_new_coordinator=[True, False],
@matrix(from_kafka_version=[str(LATEST_3_1), str(LATEST_3_2), str(LATEST_3_3), str(LATEST_3_4), str(LATEST_3_5), str(LATEST_3_6), str(LATEST_3_7), str(DEV_BRANCH)],
metadata_quorum=[isolated_kraft])
@matrix(from_kafka_version=[str(LATEST_3_7), str(DEV_BRANCH)],
use_new_coordinator=[True],
metadata_quorum=[isolated_kraft],
group_protocol=consumer_group.all_group_protocols)
def test_isolated_mode_upgrade(self, from_kafka_version, metadata_quorum, use_new_coordinator=False, group_protocol=None):
self.run_upgrade(from_kafka_version, group_protocol)
def test_isolated_mode_upgrade(self, from_kafka_version, metadata_quorum, use_new_coordinator=False):
self.run_upgrade(from_kafka_version)