Skip to content

KAFKA-16438: Update consumer_test.py’s static tests to support KIP-848’s group protocol config #15627

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wcarlson5 merged 1 commit into from
Apr 3, 2024
Merged

KAFKA-16438: Update consumer_test.py’s static tests to support KIP-848’s group protocol config #15627

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 16 additions & 13 deletions tests/kafkatest/tests/client/consumer_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -194,7 +194,7 @@ def test_consumer_bounce(self, clean_shutdown, bounce_mode, metadata_quorum=quor
static_membership=[True, False],
bounce_mode=["all", "rolling"],
num_bounces=[5],
metadata_quorum=[quorum.zk],
metadata_quorum=[quorum.zk, quorum.isolated_kraft],
use_new_coordinator=[False]
)
@matrix(
Expand All @@ -203,9 +203,10 @@ def test_consumer_bounce(self, clean_shutdown, bounce_mode, metadata_quorum=quor
bounce_mode=["all", "rolling"],
num_bounces=[5],
metadata_quorum=[quorum.isolated_kraft],
use_new_coordinator=[True, False]
use_new_coordinator=[True],
group_protocol=consumer_group.all_group_protocols
)
def test_static_consumer_bounce(self, clean_shutdown, static_membership, bounce_mode, num_bounces, metadata_quorum=quorum.zk, use_new_coordinator=False):
def test_static_consumer_bounce(self, clean_shutdown, static_membership, bounce_mode, num_bounces, metadata_quorum=quorum.zk, use_new_coordinator=False, group_protocol=None):
"""
Verify correct static consumer behavior when the consumers in the group are restarted. In order to make
sure the behavior of static members are different from dynamic ones, we take both static and dynamic
Expand All @@ -226,7 +227,7 @@ def test_static_consumer_bounce(self, clean_shutdown, static_membership, bounce_
self.await_produced_messages(producer)

self.session_timeout_sec = 60
consumer = self.setup_consumer(self.TOPIC, static_membership=static_membership)
consumer = self.setup_consumer(self.TOPIC, static_membership=static_membership, group_protocol=group_protocol)

consumer.start()
self.await_all_members(consumer)
Expand Down Expand Up @@ -268,15 +269,16 @@ def test_static_consumer_bounce(self, clean_shutdown, static_membership, bounce_
@cluster(num_nodes=7)
@matrix(
bounce_mode=["all", "rolling"],
metadata_quorum=[quorum.zk],
metadata_quorum=[quorum.zk, quorum.isolated_kraft],
use_new_coordinator=[False]
)
@matrix(
bounce_mode=["all", "rolling"],
metadata_quorum=[quorum.isolated_kraft],
use_new_coordinator=[True, False]
use_new_coordinator=[True],
group_protocol=consumer_group.all_group_protocols
)
def test_static_consumer_persisted_after_rejoin(self, bounce_mode, metadata_quorum=quorum.zk, use_new_coordinator=False):
def test_static_consumer_persisted_after_rejoin(self, bounce_mode, metadata_quorum=quorum.zk, use_new_coordinator=False, group_protocol=None):
"""
Verify that the updated member.id(updated_member_id) caused by static member rejoin would be persisted. If not,
after the brokers rolling bounce, the migrated group coordinator would load the stale persisted member.id and
Expand All @@ -291,7 +293,7 @@ def test_static_consumer_persisted_after_rejoin(self, bounce_mode, metadata_quor
producer.start()
self.await_produced_messages(producer)
self.session_timeout_sec = 60
consumer = self.setup_consumer(self.TOPIC, static_membership=True)
consumer = self.setup_consumer(self.TOPIC, static_membership=True, group_protocol=group_protocol)
consumer.start()
self.await_all_members(consumer)

Expand All @@ -309,16 +311,17 @@ def test_static_consumer_persisted_after_rejoin(self, bounce_mode, metadata_quor
@matrix(
num_conflict_consumers=[1, 2],
fencing_stage=["stable", "all"],
metadata_quorum=[quorum.zk],
metadata_quorum=[quorum.zk, quorum.isolated_kraft],
use_new_coordinator=[False]
)
@matrix(
num_conflict_consumers=[1, 2],
fencing_stage=["stable", "all"],
metadata_quorum=[quorum.isolated_kraft],
use_new_coordinator=[True, False]
use_new_coordinator=[True],
group_protocol=consumer_group.all_group_protocols
)
def test_fencing_static_consumer(self, num_conflict_consumers, fencing_stage, metadata_quorum=quorum.zk, use_new_coordinator=False):
def test_fencing_static_consumer(self, num_conflict_consumers, fencing_stage, metadata_quorum=quorum.zk, use_new_coordinator=False, group_protocol=None):
"""
Verify correct static consumer behavior when there are conflicting consumers with same group.instance.id.

Expand All @@ -335,10 +338,10 @@ def test_fencing_static_consumer(self, num_conflict_consumers, fencing_stage, me
self.await_produced_messages(producer)

self.session_timeout_sec = 60
consumer = self.setup_consumer(self.TOPIC, static_membership=True)
consumer = self.setup_consumer(self.TOPIC, static_membership=True, group_protocol=group_protocol)

self.num_consumers = num_conflict_consumers
conflict_consumer = self.setup_consumer(self.TOPIC, static_membership=True)
conflict_consumer = self.setup_consumer(self.TOPIC, static_membership=True, group_protocol=group_protocol)

# wait original set of consumer to stable stage before starting conflict members.
if fencing_stage == "stable":
Expand Down