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

KAFKA-16271: Upgrade consumer_rolling_upgrade_test.py (#15578) 

Upgrading the test to use the consumer group protocol. The two tests are failing due to Mismatch Assignment

Reviewers: Lucas Brutschy <lbrutschy@confluent.io>
This commit is contained in:
Philip Nee 2024-03-22 03:17:49 -07:00 committed by GitHub
parent 997ca14f80
commit fb03da0df4
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 8 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
tests/kafkatest/tests/client/consumer_rolling_upgrade_test.py
View File

@ -18,7 +18,7 @@ from ducktape.mark.resource import cluster
from kafkatest.tests.verifiable_consumer_test import VerifiableConsumerTest
from kafkatest.services.kafka import TopicPartition, quorum
from kafkatest.services.kafka import TopicPartition, quorum, consumer_group
class ConsumerRollingUpgradeTest(VerifiableConsumerTest):
TOPIC = "test_topic"
@ -56,7 +56,12 @@ class ConsumerRollingUpgradeTest(VerifiableConsumerTest):
metadata_quorum=[quorum.isolated_kraft],
use_new_coordinator=[True, False]
)
def rolling_update_test(self, metadata_quorum=quorum.zk, use_new_coordinator=False):
@matrix(
metadata_quorum=quorum.all_kraft,
use_new_coordinator=[True],
group_protocol=consumer_group.all_group_protocols
)
def rolling_update_test(self, metadata_quorum=quorum.zk, use_new_coordinator=False, group_protocol=None):
"""
Verify rolling updates of partition assignment strategies works correctly. In this
test, we use a rolling restart to change the group's assignment strategy from "range"
@ -65,7 +70,7 @@ class ConsumerRollingUpgradeTest(VerifiableConsumerTest):
"""
# initialize the consumer using range assignment
consumer = self.setup_consumer(self.TOPIC, assignment_strategy=self.RANGE)
consumer = self.setup_consumer(self.TOPIC, assignment_strategy=self.RANGE, group_protocol=group_protocol)
consumer.start()
self.await_all_members(consumer)