Skip to content

MINOR: replace deprecated exactly_once_beta into exactly_once_v2 #10884

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
cadonna merged 5 commits into from
Sep 27, 2021
Merged

MINOR: replace deprecated exactly_once_beta into exactly_once_v2 #10884

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
c678020
replace deprecated exactly_once_beta into exactly_once_v2
showuon Jun 15, 2021
364827a
use exactly_once_v2
showuon Jun 22, 2021
d458654
Merge branch 'trunk' of https://github.com/apache/kafka into systemTe…
showuon Jul 19, 2021
0d0718c
remove unreleased LATEST_3_0
showuon Jul 19, 2021
2bc5667
Merge branch 'trunk' of https://github.com/apache/kafka into systemTe…
showuon Sep 27, 2021
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
47 changes: 25 additions & 22 deletions tests/kafkatest/tests/streams/streams_broker_compatibility_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,17 +22,18 @@
from kafkatest.services.verifiable_consumer import VerifiableConsumer
from kafkatest.services.zookeeper import ZookeeperService
from kafkatest.version import LATEST_0_11_0, LATEST_0_10_2, LATEST_0_10_1, LATEST_0_10_0, LATEST_1_0, LATEST_1_1, \
LATEST_2_0, LATEST_2_1, LATEST_2_2, LATEST_2_3, LATEST_2_4, LATEST_2_5, LATEST_2_6, KafkaVersion
LATEST_2_0, LATEST_2_1, LATEST_2_2, LATEST_2_3, LATEST_2_4, LATEST_2_5, LATEST_2_6, LATEST_2_7, LATEST_2_8, \
KafkaVersion


class StreamsBrokerCompatibility(Test):
"""
These tests validates that
- Streams works for older brokers 0.11 (or newer)
- Streams w/ EOS-alpha works for older brokers 0.11 (or newer)
- Streams w/ EOS-beta works for older brokers 2.5 (or newer)
- Streams w/ EOS-v2 works for older brokers 2.5 (or newer)
- Streams fails fast for older brokers 0.10.0, 0.10.2, and 0.10.1
- Streams w/ EOS-beta fails fast for older brokers 2.4 or older
- Streams w/ EOS-v2 fails fast for older brokers 2.4 or older
"""

input = "brokerCompatibilitySourceTopic"
Expand Down Expand Up @@ -114,23 +115,25 @@ def test_compatible_brokers_eos_alpha_enabled(self, broker_version):
self.consumer.stop()
self.kafka.stop()

# TODO enable after 2.5 is released
# @parametrize(broker_version=str(LATEST_2_5))
# def test_compatible_brokers_eos_beta_enabled(self, broker_version):

@showuon showuon Jun 15, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We forgot to enable it after V2.5. Enable it now.

# self.kafka.set_version(KafkaVersion(broker_version))
# self.kafka.start()
#
# processor = StreamsBrokerCompatibilityService(self.test_context, self.kafka, "exactly_once_beta")
# processor.start()
#
# self.consumer.start()
#
# processor.wait()
#
# wait_until(lambda: self.consumer.total_consumed() > 0, timeout_sec=30, err_msg="Did expect to read a message but got none within 30 seconds.")
#
# self.consumer.stop()
# self.kafka.stop()
@parametrize(broker_version=str(LATEST_2_8))
@parametrize(broker_version=str(LATEST_2_7))
@parametrize(broker_version=str(LATEST_2_6))
@parametrize(broker_version=str(LATEST_2_5))
def test_compatible_brokers_eos_v2_enabled(self, broker_version):
self.kafka.set_version(KafkaVersion(broker_version))
self.kafka.start()

processor = StreamsBrokerCompatibilityService(self.test_context, self.kafka, "exactly_once_v2")
processor.start()

self.consumer.start()

processor.wait()

wait_until(lambda: self.consumer.total_consumed() > 0, timeout_sec=30, err_msg="Did expect to read a message but got none within 30 seconds.")

self.consumer.stop()
self.kafka.stop()

@cluster(num_nodes=4)
@parametrize(broker_version=str(LATEST_0_10_2))
Expand Down Expand Up @@ -159,11 +162,11 @@ def test_fail_fast_on_incompatible_brokers(self, broker_version):
@parametrize(broker_version=str(LATEST_1_1))
@parametrize(broker_version=str(LATEST_1_0))
@parametrize(broker_version=str(LATEST_0_11_0))
def test_fail_fast_on_incompatible_brokers_if_eos_beta_enabled(self, broker_version):
def test_fail_fast_on_incompatible_brokers_if_eos_v2_enabled(self, broker_version):
self.kafka.set_version(KafkaVersion(broker_version))
self.kafka.start()

processor = StreamsBrokerCompatibilityService(self.test_context, self.kafka, "exactly_once_beta")
processor = StreamsBrokerCompatibilityService(self.test_context, self.kafka, "exactly_once_v2")

with processor.node.account.monitor_log(processor.STDERR_FILE) as monitor:
with processor.node.account.monitor_log(processor.LOG_FILE) as log:
Expand Down
8 changes: 4 additions & 4 deletions tests/kafkatest/tests/streams/streams_eos_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ def __init__(self, test_context):

@cluster(num_nodes=9)
@parametrize(processing_guarantee="exactly_once")
@parametrize(processing_guarantee="exactly_once_beta")
@parametrize(processing_guarantee="exactly_once_v2")
def test_rebalance_simple(self, processing_guarantee):
self.run_rebalance(StreamsEosTestJobRunnerService(self.test_context, self.kafka, processing_guarantee),
StreamsEosTestJobRunnerService(self.test_context, self.kafka, processing_guarantee),
Expand All @@ -48,7 +48,7 @@ def test_rebalance_simple(self, processing_guarantee):

@cluster(num_nodes=9)
@parametrize(processing_guarantee="exactly_once")
@parametrize(processing_guarantee="exactly_once_beta")
@parametrize(processing_guarantee="exactly_once_v2")
def test_rebalance_complex(self, processing_guarantee):
self.run_rebalance(StreamsComplexEosTestJobRunnerService(self.test_context, self.kafka, processing_guarantee),
StreamsComplexEosTestJobRunnerService(self.test_context, self.kafka, processing_guarantee),
Expand Down Expand Up @@ -83,7 +83,7 @@ def run_rebalance(self, processor1, processor2, processor3, verifier):

@cluster(num_nodes=9)
@parametrize(processing_guarantee="exactly_once")
@parametrize(processing_guarantee="exactly_once_beta")
@parametrize(processing_guarantee="exactly_once_v2")
def test_failure_and_recovery(self, processing_guarantee):
self.run_failure_and_recovery(StreamsEosTestJobRunnerService(self.test_context, self.kafka, processing_guarantee),
StreamsEosTestJobRunnerService(self.test_context, self.kafka, processing_guarantee),
Expand All @@ -92,7 +92,7 @@ def test_failure_and_recovery(self, processing_guarantee):

@cluster(num_nodes=9)
@parametrize(processing_guarantee="exactly_once")
@parametrize(processing_guarantee="exactly_once_beta")
@parametrize(processing_guarantee="exactly_once_v2")
def test_failure_and_recovery_complex(self, processing_guarantee):
self.run_failure_and_recovery(StreamsComplexEosTestJobRunnerService(self.test_context, self.kafka, processing_guarantee),
StreamsComplexEosTestJobRunnerService(self.test_context, self.kafka, processing_guarantee),
Expand Down
4 changes: 2 additions & 2 deletions tests/kafkatest/tests/streams/streams_relational_smoke_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,11 +85,11 @@ def __init__(self, test_context):

@cluster(num_nodes=8)
@matrix(crash=[False, True],
processing_guarantee=['exactly_once', 'exactly_once_beta'])
processing_guarantee=['exactly_once', 'exactly_once_v2'])
def test_streams(self, crash, processing_guarantee):
driver = StreamsRelationalSmokeTestService(self.test_context, self.kafka, "driver", "ignored", "ignored")

LOG_FILE = driver.LOG_FILE # this is the same for all instaces of the service, so we can just declare a "constant"
LOG_FILE = driver.LOG_FILE # this is the same for all instances of the service, so we can just declare a "constant"

processor1 = StreamsRelationalSmokeTestService(self.test_context, self.kafka, "application", "processor1", processing_guarantee)
processor2 = StreamsRelationalSmokeTestService(self.test_context, self.kafka, "application", "processor2", processing_guarantee)
Expand Down