lucasbru commented on code in PR #23523:
URL: https://github.com/apache/kafka/pull/23523#discussion_r4072951134
##########
tests/kafkatest/tests/streams/streams_static_membership_test.py:
##########
@@ -361,6 +362,53 @@ def test_fencing_static_streams_member(self,
fencing_stage, metadata_quorum):
self.producer.stop()
self.kafka.stop(timeout_sec=120)
+ @cluster(num_nodes=8)
+ @matrix(
+ broker_version=[str(LATEST_4_2), str(LATEST_4_3)],
Review Comment:
Seems like one of the two would be enough?
##########
tests/kafkatest/tests/streams/streams_static_membership_test.py:
##########
@@ -361,6 +362,53 @@ def test_fencing_static_streams_member(self,
fencing_stage, metadata_quorum):
self.producer.stop()
self.kafka.stop(timeout_sec=120)
+ @cluster(num_nodes=8)
+ @matrix(
+ broker_version=[str(LATEST_4_2), str(LATEST_4_3)],
+ metadata_quorum=[quorum.isolated_kraft]
+ )
+ def test_new_streams_client_static_membership_against_old_broker(
+ self, broker_version, metadata_quorum):
+ """
+ KAFKA-20971: a newer Kafka Streams client (dev/4.4+) using static
membership on the
+ `streams` group protocol should fail with a clear unsupported-version
error against
+ an older (4.2/4.3) broker that doesn't support the `streams` group
protocol.
+ """
+ self.kafka.set_version(KafkaVersion(broker_version))
+ if self.kafka.isolated_controller_quorum:
+
self.kafka.isolated_controller_quorum.set_version(KafkaVersion(broker_version))
+ self.kafka.start()
+
+ processor = StaticMemberTestService(
+ self.test_context,
+ self.kafka,
+ "consumer-A",
+ self.num_threads,
+ self.streams_group_protocol
+ )
+
+ self.set_topics(processor)
+
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.start_node(processor.node)
+
+ monitor.wait_until(
+ "Static membership is not yet supported",
+ timeout_sec=60,
+ err_msg="Streams client did not report the expected
static-membership-unsupported "
+ "error against a %s broker" %
broker_version
+ )
+
+ monitor.wait_until(
+ "transitioned to the terminal ERROR state",
+ timeout_sec=60,
+ err_msg="Streams client did not shut down to ERROR
state after rejecting "
+ "static membership against a %s broker" %
broker_version
+ )
+
+ processor.stop_node(processor.node, clean_shutdown=False)
Review Comment:
Since the point of the test is that the client reaches a terminal state
instead of hanging/retrying, could we call `processor.wait(timeout_sec=60)`
here (like `conflict_processor.wait(...)` in the fencing test) before the
forced stop_node? As written we only check that the ERROR-state message was
logged once, not that the process actually exits.
##########
tests/kafkatest/tests/streams/streams_static_membership_test.py:
##########
@@ -361,6 +362,53 @@ def test_fencing_static_streams_member(self,
fencing_stage, metadata_quorum):
self.producer.stop()
self.kafka.stop(timeout_sec=120)
+ @cluster(num_nodes=8)
+ @matrix(
+ broker_version=[str(LATEST_4_2), str(LATEST_4_3)],
+ metadata_quorum=[quorum.isolated_kraft]
+ )
+ def test_new_streams_client_static_membership_against_old_broker(
+ self, broker_version, metadata_quorum):
+ """
+ KAFKA-20971: a newer Kafka Streams client (dev/4.4+) using static
membership on the
+ `streams` group protocol should fail with a clear unsupported-version
error against
+ an older (4.2/4.3) broker that doesn't support the `streams` group
protocol.
+ """
+ self.kafka.set_version(KafkaVersion(broker_version))
+ if self.kafka.isolated_controller_quorum:
+
self.kafka.isolated_controller_quorum.set_version(KafkaVersion(broker_version))
+ self.kafka.start()
+
+ processor = StaticMemberTestService(
+ self.test_context,
+ self.kafka,
+ "consumer-A",
+ self.num_threads,
+ self.streams_group_protocol
+ )
+
+ self.set_topics(processor)
+
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.start_node(processor.node)
+
+ monitor.wait_until(
+ "Static membership is not yet supported",
+ timeout_sec=60,
+ err_msg="Streams client did not report the expected
static-membership-unsupported "
+ "error against a %s broker" %
broker_version
+ )
+
+ monitor.wait_until(
+ "transitioned to the terminal ERROR state",
Review Comment:
This exact string is only logged from `close()`'s `isShuttingDown()` branch,
when a second caller races the async shutdown thread and has to wait for it.
The autonomous fatal path this test hits (`closeToError()` -> daemon shutdown
thread -> `setState(ERROR)`) never logs it directly - it only shows up if the
client's JVM shutdown hook happens to race that daemon thread's `close()`
before it finishes. If the state is already ERROR by the time the hook runs,
you get "already in the terminal ERROR state" instead, which doesn't match this
substring and the assertion would time out.
`test_fencing_static_streams_member` above uses the looser `"terminal ERROR
state"` for what looks like this exact reason - might be safer to match that
here too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
