This is an automated email from the ASF dual-hosted git repository.
bbejeck pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 0c2c0df04a3 KAFKA-20727: Update system tests for transactional state
stores (#22642)
0c2c0df04a3 is described below
commit 0c2c0df04a3ac7390aaebdf9484c6b49c7f883fc
Author: Bill Bejeck <[email protected]>
AuthorDate: Fri Jun 26 09:05:17 2026 -0400
KAFKA-20727: Update system tests for transactional state stores (#22642)
Adds transactional state store coverage to the Streams smoke test.
`StreamsSmokeTestJobRunnerService` (and its base) take a new
transactional flag that sets `enable.transactional.statestores=true` in
`streams.properties`; the config flows through to StreamsConfig with no
Java changes.
This exercises the transactional RocksDB and in-memory stores under EOS
with hard crashes
Reviewers: Matthias J. Sax <[email protected]>
---
tests/kafkatest/services/streams.py | 10 +++++++---
tests/kafkatest/tests/streams/streams_smoke_test.py | 15 +++++++++++----
2 files changed, 18 insertions(+), 7 deletions(-)
diff --git a/tests/kafkatest/services/streams.py
b/tests/kafkatest/services/streams.py
index 49ba32ad8e1..c387a5e20d4 100644
--- a/tests/kafkatest/services/streams.py
+++ b/tests/kafkatest/services/streams.py
@@ -323,7 +323,7 @@ class StreamsTestBaseService(KafkaPathResolverMixin,
JmxMixin, Service):
class StreamsSmokeTestBaseService(StreamsTestBaseService):
"""Base class for Streams Smoke Test services providing some common
settings and functionality"""
- def __init__(self, test_context, kafka, command, processing_guarantee =
'at_least_once', group_protocol = 'classic', num_threads = 3,
replication_factor = 3):
+ def __init__(self, test_context, kafka, command, processing_guarantee =
'at_least_once', group_protocol = 'classic', num_threads = 3,
replication_factor = 3, transactional = False):
super(StreamsSmokeTestBaseService, self).__init__(test_context,
kafka,
"org.apache.kafka.streams.tests.StreamsSmokeTest",
@@ -334,6 +334,7 @@ class StreamsSmokeTestBaseService(StreamsTestBaseService):
self.KAFKA_STREAMS_VERSION = ""
self.UPGRADE_FROM = None
self.REPLICATION_FACTOR = replication_factor
+ self.TRANSACTIONAL = transactional
def set_version(self, kafka_streams_version):
self.KAFKA_STREAMS_VERSION = kafka_streams_version
@@ -360,6 +361,9 @@ class StreamsSmokeTestBaseService(StreamsTestBaseService):
if self.UPGRADE_FROM is not None:
properties['upgrade.from'] = self.UPGRADE_FROM
+ if self.TRANSACTIONAL:
+ properties['enable.transactional.statestores'] = "true"
+
cfg = KafkaConfig(**properties)
return cfg.render()
@@ -414,8 +418,8 @@ class
StreamsSmokeTestDriverService(StreamsSmokeTestBaseService):
return cmd
class StreamsSmokeTestJobRunnerService(StreamsSmokeTestBaseService):
- def __init__(self, test_context, kafka, processing_guarantee,
group_protocol = 'classic', num_threads = 3, replication_factor = 3):
- super(StreamsSmokeTestJobRunnerService, self).__init__(test_context,
kafka, "process", processing_guarantee, group_protocol, num_threads,
replication_factor)
+ def __init__(self, test_context, kafka, processing_guarantee,
group_protocol = 'classic', num_threads = 3, replication_factor = 3,
transactional = False):
+ super(StreamsSmokeTestJobRunnerService, self).__init__(test_context,
kafka, "process", processing_guarantee, group_protocol, num_threads,
replication_factor, transactional)
class StreamsSmokeTestShutdownDeadlockService(StreamsSmokeTestBaseService):
def __init__(self, test_context, kafka):
diff --git a/tests/kafkatest/tests/streams/streams_smoke_test.py
b/tests/kafkatest/tests/streams/streams_smoke_test.py
index e74c400d861..190a28c84e5 100644
--- a/tests/kafkatest/tests/streams/streams_smoke_test.py
+++ b/tests/kafkatest/tests/streams/streams_smoke_test.py
@@ -57,10 +57,17 @@ class StreamsSmokeTest(BaseStreamsTest):
metadata_quorum=[quorum.combined_kraft],
group_protocol=["streams"],
enable_assignment_batching=[False, True])
- def test_streams(self, processing_guarantee, crash, metadata_quorum,
group_protocol, enable_assignment_batching):
- processor1 = StreamsSmokeTestJobRunnerService(self.test_context,
self.kafka, processing_guarantee, group_protocol)
- processor2 = StreamsSmokeTestJobRunnerService(self.test_context,
self.kafka, processing_guarantee, group_protocol)
- processor3 = StreamsSmokeTestJobRunnerService(self.test_context,
self.kafka, processing_guarantee, group_protocol)
+ # transactional state stores (KIP-892) are an EOS feature; exercise them
under EOS + crash
+ @matrix(processing_guarantee=['exactly_once_v2'],
+ crash=[True, False],
+ metadata_quorum=[quorum.combined_kraft],
+ group_protocol=["streams"],
+ enable_assignment_batching=[True],
+ transactional=[True])
+ def test_streams(self, processing_guarantee, crash, metadata_quorum,
group_protocol, enable_assignment_batching, transactional=False):
+ processor1 = StreamsSmokeTestJobRunnerService(self.test_context,
self.kafka, processing_guarantee, group_protocol, transactional=transactional)
+ processor2 = StreamsSmokeTestJobRunnerService(self.test_context,
self.kafka, processing_guarantee, group_protocol, transactional=transactional)
+ processor3 = StreamsSmokeTestJobRunnerService(self.test_context,
self.kafka, processing_guarantee, group_protocol, transactional=transactional)
with processor1.node.account.monitor_log(processor1.STDOUT_FILE) as
monitor1:
processor1.start()
