You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lucliu1108 (via GitHub)" <gi...@apache.org> on 2026/08/12 04:31:54 UTC
[PR] KAFKA-20817: Add more ducktape system tests scenarios for topology description plugin [kafka]
lucliu1108 opened a new pull request, #23138:
URL: https://github.com/apache/kafka/pull/23138
## Summary
This PR adds 3 more system test scenarios for streams topology test.
1. `test_topology_description_not_resolicited_after_client_restart`: client start with description already stored, the broker is already running -> client restarted and since `storedDescriptionTopologyEpoch == currentTopologyEpoch`, the broker does not solicit again.
2. `test_topology_description_only_one_member_pushes`: two members join the same streams group at the same time and both heartbeat to the broker -> since `StreamsGroupTopologyDescriptionBackoff.armIfNotActive`'s check-and-arm on the group's back-off entry is atomic, only one member's heartbeat gets `TopologyDescriptionRequired=true`; the other member never sends a push.
3. `test_topology_description_resolicited_after_group_delete_and_recreate`: client pushes successfully, then the group is stopped and deleted -> a new client joins under the same application.id and since the delete clears the old group's `storedDescriptionTopologyEpoch`, the broker treats it as a brand-new group and solicits again.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20817: Add more ducktape system tests scenarios for topology description plugin [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #23138:
URL: https://github.com/apache/kafka/pull/23138#discussion_r3829844233
##########
tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py:
##########
@@ -161,3 +165,151 @@ def test_topology_description_not_stored_without_plugin(self, metadata_quorum):
assert int(next(pushed).strip()) == 0, \
"Client logged a successful push despite no plugin being configured on the broker"
processor.stop()
+
+ @cluster(num_nodes=2)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_not_resolicited_after_client_restart(self, metadata_quorum):
+ """
+ Test the situation when the client restarts after already having pushed its
+ topology description successfully. The broker still has storedDescriptionTopologyEpoch
+ matching currentTopologyEpoch, so it must not solicit a second push.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ processor = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.start()
+ monitor.wait_until(self.PUSH_SUCCESS_LOG,
+ timeout_sec=120,
+ err_msg="Streams client did not log a successful topology description push")
+
+ broker_node = self.kafka.nodes[0]
+ solicited_before = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ solicited_before_count = int(next(solicited_before).strip())
+ assert solicited_before_count > 0, \
+ "Broker never solicited the initial topology push despite the plugin being configured"
+
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.restart()
+ monitor.wait_until(self.STREAMS_RUNNING_LOG,
+ timeout_sec=60,
+ err_msg="Never saw 'REBALANCING -> RUNNING' message after client restart " + str(processor.node.account))
+
+ solicited_after = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ assert int(next(solicited_after).strip()) == solicited_before_count, \
+ "Broker re-solicited a topology push after a client restart despite an already-stored, matching-epoch description"
+
+ time.sleep(5)
+
+ # Check after a default heartbeat interval that broker still doesn't re-solicit a push
+ solicited_settled = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ assert int(next(solicited_settled).strip()) == solicited_before_count, \
+ "Broker re-solicited a topology push on a later heartbeat after a client restart despite an already-stored, matching-epoch description"
+
+ pushed = processor.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor.LOG_FILE),
+ allow_fail=False)
+ assert int(next(pushed).strip()) == 1, \
+ "Client pushed a topology description again after restart despite the broker not soliciting"
+ processor.stop()
+
+ @cluster(num_nodes=3)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_only_one_member_pushes(self, metadata_quorum):
+ """
+ Test the situation when two members of the same streams group start up together.
+ StreamsGroupTopologyDescriptionBackoff.armIfNotActive must prevent every member
+ from pushing the same description; only one member's push should succeed,
+ regardless of which member wins the race.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ processor1 = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ processor2 = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+
Review Comment:
Trailing whitespace on this blank line (shows as a red mark in the diff) — worth clearing. Minor.
##########
tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py:
##########
@@ -28,6 +31,7 @@ class StreamsTopologyDescriptionPluginTest(Test):
PUSH_REQUESTED_LOG = "Broker requested topology description push"
PUSH_SENDING_LOG = "Sending topology description for group"
PUSH_SUCCESS_LOG = "Topology description pushed successfully"
+ PUSH_FAILED_LOG = "Topology description push failed with non-retriable exception"
Review Comment:
`PUSH_FAILED_LOG` is now unused — its only reader, `total_push_failures()`, was removed in the last commit. Drop the constant.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20817: Add more ducktape system tests scenarios for topology description plugin [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #23138:
URL: https://github.com/apache/kafka/pull/23138#discussion_r3817191710
##########
tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py:
##########
@@ -161,3 +163,143 @@ def test_topology_description_not_stored_without_plugin(self, metadata_quorum):
assert int(next(pushed).strip()) == 0, \
"Client logged a successful push despite no plugin being configured on the broker"
processor.stop()
+
+ @cluster(num_nodes=2)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_not_resolicited_after_client_restart(self, metadata_quorum):
+ """
+ Test the situation when the client restarts after already having pushed its
+ topology description successfully. The broker still has storedDescriptionTopologyEpoch
+ matching currentTopologyEpoch, so it must not solicit a second push.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ processor = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.start()
+ monitor.wait_until(self.PUSH_SUCCESS_LOG,
+ timeout_sec=120,
+ err_msg="Streams client did not log a successful topology description push")
+
+ broker_node = self.kafka.nodes[0]
+ solicited_before = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ solicited_before_count = int(next(solicited_before).strip())
+ assert solicited_before_count > 0, \
+ "Broker never solicited the initial topology push despite the plugin being configured"
+
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.restart()
+ monitor.wait_until(self.STREAMS_RUNNING_LOG,
+ timeout_sec=60,
+ err_msg="Never saw 'REBALANCING -> RUNNING' message after client restart " + str(processor.node.account))
+
+ solicited_after = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ assert int(next(solicited_after).strip()) == solicited_before_count, \
+ "Broker re-solicited a topology push after a client restart despite an already-stored, matching-epoch description"
+
+ pushed = processor.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor.LOG_FILE),
+ allow_fail=False)
+ assert int(next(pushed).strip()) == 1, \
+ "Client pushed a topology description again after restart despite the broker not soliciting"
+ processor.stop()
+
+ @cluster(num_nodes=3)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_only_one_member_pushes(self, metadata_quorum):
+ """
+ Test the situation when two members of the same streams group start up together.
+ StreamsGroupTopologyDescriptionManager.armIfNotActive must prevent every member
+ from pushing the same description; only one member's push should succeed,
+ regardless of which member wins the race.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ processor1 = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ processor2 = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ processor1.start()
+ processor2.start()
+
+ def total_push_successes():
+ pushed1 = processor1.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor1.LOG_FILE),
+ allow_fail=False)
+ pushed2 = processor2.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor2.LOG_FILE),
+ allow_fail=False)
+ return int(next(pushed1).strip()) + int(next(pushed2).strip())
+
+ def total_push_failures():
+ failed1 = processor1.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_FAILED_LOG, processor1.LOG_FILE),
+ allow_fail=False)
+ failed2 = processor2.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_FAILED_LOG, processor2.LOG_FILE),
+ allow_fail=False)
+ return int(next(failed1).strip()) + int(next(failed2).strip())
+
+ wait_until(lambda: total_push_successes() >= 1,
Review Comment:
Fix: for `test_topology_description_only_one_member_pushes`,
- wait until 2 processors all reach RUNNING state, so it prevents the situation where one member fails to run while the other pushes successfully
- check `push_successes==1 && total_sent==1` instead of checking on push failure counts.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20817: Add more ducktape system tests scenarios for topology description plugin [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #23138:
URL: https://github.com/apache/kafka/pull/23138#discussion_r3817344010
##########
tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py:
##########
@@ -161,3 +163,143 @@ def test_topology_description_not_stored_without_plugin(self, metadata_quorum):
assert int(next(pushed).strip()) == 0, \
"Client logged a successful push despite no plugin being configured on the broker"
processor.stop()
+
+ @cluster(num_nodes=2)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_not_resolicited_after_client_restart(self, metadata_quorum):
+ """
+ Test the situation when the client restarts after already having pushed its
+ topology description successfully. The broker still has storedDescriptionTopologyEpoch
+ matching currentTopologyEpoch, so it must not solicit a second push.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ processor = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.start()
+ monitor.wait_until(self.PUSH_SUCCESS_LOG,
+ timeout_sec=120,
+ err_msg="Streams client did not log a successful topology description push")
+
+ broker_node = self.kafka.nodes[0]
+ solicited_before = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ solicited_before_count = int(next(solicited_before).strip())
+ assert solicited_before_count > 0, \
+ "Broker never solicited the initial topology push despite the plugin being configured"
+
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.restart()
+ monitor.wait_until(self.STREAMS_RUNNING_LOG,
+ timeout_sec=60,
+ err_msg="Never saw 'REBALANCING -> RUNNING' message after client restart " + str(processor.node.account))
+
+ solicited_after = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ assert int(next(solicited_after).strip()) == solicited_before_count, \
+ "Broker re-solicited a topology push after a client restart despite an already-stored, matching-epoch description"
+
+ pushed = processor.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor.LOG_FILE),
+ allow_fail=False)
+ assert int(next(pushed).strip()) == 1, \
+ "Client pushed a topology description again after restart despite the broker not soliciting"
+ processor.stop()
+
+ @cluster(num_nodes=3)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_only_one_member_pushes(self, metadata_quorum):
+ """
+ Test the situation when two members of the same streams group start up together.
+ StreamsGroupTopologyDescriptionManager.armIfNotActive must prevent every member
+ from pushing the same description; only one member's push should succeed,
+ regardless of which member wins the race.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ processor1 = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ processor2 = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ processor1.start()
+ processor2.start()
+
+ def total_push_successes():
+ pushed1 = processor1.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor1.LOG_FILE),
+ allow_fail=False)
+ pushed2 = processor2.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor2.LOG_FILE),
+ allow_fail=False)
+ return int(next(pushed1).strip()) + int(next(pushed2).strip())
+
+ def total_push_failures():
+ failed1 = processor1.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_FAILED_LOG, processor1.LOG_FILE),
+ allow_fail=False)
+ failed2 = processor2.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_FAILED_LOG, processor2.LOG_FILE),
+ allow_fail=False)
+ return int(next(failed1).strip()) + int(next(failed2).strip())
+
+ wait_until(lambda: total_push_successes() >= 1,
+ timeout_sec=120,
+ err_msg=lambda: "Neither streams client logged a successful topology description push"
+ + (" (a non-retriable push failure was logged instead, see client logs)"
+ if total_push_failures() > 0 else ""))
+ assert total_push_failures() == 0, \
+ "A member logged a non-retriable push failure despite a push having succeeded"
+ assert total_push_successes() == 1, \
+ "Expected exactly one member to push the topology description successfully"
+
+ sent1 = processor1.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SENDING_LOG, processor1.LOG_FILE),
+ allow_fail=False)
+ sent2 = processor2.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SENDING_LOG, processor2.LOG_FILE),
+ allow_fail=False)
+ total_sent = int(next(sent1).strip()) + int(next(sent2).strip())
+ assert total_sent == 1, \
+ "Expected exactly one member to send a topology description, got %d" % total_sent
+
+ processor1.stop()
+ processor2.stop()
+
+ @cluster(num_nodes=3)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_resolicited_after_group_delete_and_recreate(self, metadata_quorum):
+ """
+ Test the situation when a streams group is deleted after a successful push, then a
+ new client joins under the same application.id. GroupCoordinatorShard.
+ finalizeStoredDescriptionTopologyEpochAfterDelete clears the deleted group's stored
Review Comment:
Fix: change to `GroupMetadataManager.finalizeStoredDescriptionTopologyEpochAfterDelete` in the doc since this one works on a specific group and is more accurate, while the `GroupCoordinatorShard` one work on a batch of groups.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20817: Add more ducktape system tests scenarios for topology description plugin [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #23138:
URL: https://github.com/apache/kafka/pull/23138#discussion_r3831026017
##########
tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py:
##########
@@ -28,6 +31,7 @@ class StreamsTopologyDescriptionPluginTest(Test):
PUSH_REQUESTED_LOG = "Broker requested topology description push"
PUSH_SENDING_LOG = "Sending topology description for group"
PUSH_SUCCESS_LOG = "Topology description pushed successfully"
+ PUSH_FAILED_LOG = "Topology description push failed with non-retriable exception"
Review Comment:
Fix: i removed the whitespace and the unused constant.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20817: Add more ducktape system tests scenarios for topology description plugin [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #23138:
URL: https://github.com/apache/kafka/pull/23138#discussion_r3817259660
##########
tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py:
##########
@@ -161,3 +163,143 @@ def test_topology_description_not_stored_without_plugin(self, metadata_quorum):
assert int(next(pushed).strip()) == 0, \
"Client logged a successful push despite no plugin being configured on the broker"
processor.stop()
+
+ @cluster(num_nodes=2)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_not_resolicited_after_client_restart(self, metadata_quorum):
+ """
+ Test the situation when the client restarts after already having pushed its
+ topology description successfully. The broker still has storedDescriptionTopologyEpoch
+ matching currentTopologyEpoch, so it must not solicit a second push.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ processor = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.start()
+ monitor.wait_until(self.PUSH_SUCCESS_LOG,
+ timeout_sec=120,
+ err_msg="Streams client did not log a successful topology description push")
+
+ broker_node = self.kafka.nodes[0]
+ solicited_before = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ solicited_before_count = int(next(solicited_before).strip())
+ assert solicited_before_count > 0, \
+ "Broker never solicited the initial topology push despite the plugin being configured"
+
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.restart()
+ monitor.wait_until(self.STREAMS_RUNNING_LOG,
+ timeout_sec=60,
+ err_msg="Never saw 'REBALANCING -> RUNNING' message after client restart " + str(processor.node.account))
+
+ solicited_after = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ assert int(next(solicited_after).strip()) == solicited_before_count, \
+ "Broker re-solicited a topology push after a client restart despite an already-stored, matching-epoch description"
+
+ pushed = processor.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor.LOG_FILE),
+ allow_fail=False)
+ assert int(next(pushed).strip()) == 1, \
+ "Client pushed a topology description again after restart despite the broker not soliciting"
+ processor.stop()
+
+ @cluster(num_nodes=3)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_only_one_member_pushes(self, metadata_quorum):
+ """
+ Test the situation when two members of the same streams group start up together.
+ StreamsGroupTopologyDescriptionManager.armIfNotActive must prevent every member
Review Comment:
Fixed the doc to `StreamsGroupTopologyDescriptionBackoff.armIfNotActive`
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20817: Add more ducktape system tests scenarios for topology description plugin [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #23138:
URL: https://github.com/apache/kafka/pull/23138#discussion_r3817262790
##########
tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py:
##########
@@ -161,3 +163,143 @@ def test_topology_description_not_stored_without_plugin(self, metadata_quorum):
assert int(next(pushed).strip()) == 0, \
"Client logged a successful push despite no plugin being configured on the broker"
processor.stop()
+
+ @cluster(num_nodes=2)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_not_resolicited_after_client_restart(self, metadata_quorum):
+ """
+ Test the situation when the client restarts after already having pushed its
+ topology description successfully. The broker still has storedDescriptionTopologyEpoch
+ matching currentTopologyEpoch, so it must not solicit a second push.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ processor = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.start()
+ monitor.wait_until(self.PUSH_SUCCESS_LOG,
+ timeout_sec=120,
+ err_msg="Streams client did not log a successful topology description push")
+
+ broker_node = self.kafka.nodes[0]
+ solicited_before = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ solicited_before_count = int(next(solicited_before).strip())
+ assert solicited_before_count > 0, \
+ "Broker never solicited the initial topology push despite the plugin being configured"
+
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.restart()
+ monitor.wait_until(self.STREAMS_RUNNING_LOG,
+ timeout_sec=60,
+ err_msg="Never saw 'REBALANCING -> RUNNING' message after client restart " + str(processor.node.account))
+
+ solicited_after = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ assert int(next(solicited_after).strip()) == solicited_before_count, \
+ "Broker re-solicited a topology push after a client restart despite an already-stored, matching-epoch description"
+
+ pushed = processor.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor.LOG_FILE),
+ allow_fail=False)
+ assert int(next(pushed).strip()) == 1, \
+ "Client pushed a topology description again after restart despite the broker not soliciting"
+ processor.stop()
+
+ @cluster(num_nodes=3)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_only_one_member_pushes(self, metadata_quorum):
+ """
+ Test the situation when two members of the same streams group start up together.
+ StreamsGroupTopologyDescriptionManager.armIfNotActive must prevent every member
+ from pushing the same description; only one member's push should succeed,
+ regardless of which member wins the race.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ processor1 = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ processor2 = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ processor1.start()
+ processor2.start()
+
+ def total_push_successes():
+ pushed1 = processor1.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor1.LOG_FILE),
+ allow_fail=False)
+ pushed2 = processor2.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor2.LOG_FILE),
+ allow_fail=False)
+ return int(next(pushed1).strip()) + int(next(pushed2).strip())
+
+ def total_push_failures():
+ failed1 = processor1.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_FAILED_LOG, processor1.LOG_FILE),
+ allow_fail=False)
+ failed2 = processor2.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_FAILED_LOG, processor2.LOG_FILE),
+ allow_fail=False)
+ return int(next(failed1).strip()) + int(next(failed2).strip())
+
+ wait_until(lambda: total_push_successes() >= 1,
+ timeout_sec=120,
+ err_msg=lambda: "Neither streams client logged a successful topology description push"
+ + (" (a non-retriable push failure was logged instead, see client logs)"
+ if total_push_failures() > 0 else ""))
+ assert total_push_failures() == 0, \
+ "A member logged a non-retriable push failure despite a push having succeeded"
+ assert total_push_successes() == 1, \
+ "Expected exactly one member to push the topology description successfully"
+
+ sent1 = processor1.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SENDING_LOG, processor1.LOG_FILE),
+ allow_fail=False)
+ sent2 = processor2.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SENDING_LOG, processor2.LOG_FILE),
+ allow_fail=False)
+ total_sent = int(next(sent1).strip()) + int(next(sent2).strip())
+ assert total_sent == 1, \
+ "Expected exactly one member to send a topology description, got %d" % total_sent
+
+ processor1.stop()
+ processor2.stop()
+
+ @cluster(num_nodes=3)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_resolicited_after_group_delete_and_recreate(self, metadata_quorum):
+ """
+ Test the situation when a streams group is deleted after a successful push, then a
+ new client joins under the same application.id. GroupCoordinatorShard.
+ finalizeStoredDescriptionTopologyEpochAfterDelete clears the deleted group's stored
+ epoch and back-off state, so the new incarnation must be freshly solicited rather
+ than inheriting the "already stored" state left behind by the deleted group.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ group_id = "kafka-streams-system-test-topology-description-plugin"
+ processor = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.start()
+ monitor.wait_until(self.PUSH_SUCCESS_LOG,
+ timeout_sec=120,
+ err_msg="Streams client did not log a successful topology description push")
+
+ processor.stop()
+
+ def group_deleted():
+ return "was successful" in self.kafka.delete_streams_group(group_id)
+
+ wait_until(group_deleted, timeout_sec=30, backoff_sec=2,
Review Comment:
Fix: bumped the timeout to 60s.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20817: Add more ducktape system tests scenarios for topology description plugin [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #23138:
URL: https://github.com/apache/kafka/pull/23138#discussion_r3817238151
##########
tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py:
##########
@@ -161,3 +163,143 @@ def test_topology_description_not_stored_without_plugin(self, metadata_quorum):
assert int(next(pushed).strip()) == 0, \
"Client logged a successful push despite no plugin being configured on the broker"
processor.stop()
+
+ @cluster(num_nodes=2)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_not_resolicited_after_client_restart(self, metadata_quorum):
+ """
+ Test the situation when the client restarts after already having pushed its
+ topology description successfully. The broker still has storedDescriptionTopologyEpoch
+ matching currentTopologyEpoch, so it must not solicit a second push.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ processor = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.start()
+ monitor.wait_until(self.PUSH_SUCCESS_LOG,
+ timeout_sec=120,
+ err_msg="Streams client did not log a successful topology description push")
+
+ broker_node = self.kafka.nodes[0]
+ solicited_before = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ solicited_before_count = int(next(solicited_before).strip())
+ assert solicited_before_count > 0, \
+ "Broker never solicited the initial topology push despite the plugin being configured"
+
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.restart()
+ monitor.wait_until(self.STREAMS_RUNNING_LOG,
+ timeout_sec=60,
+ err_msg="Never saw 'REBALANCING -> RUNNING' message after client restart " + str(processor.node.account))
+
+ solicited_after = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ assert int(next(solicited_after).strip()) == solicited_before_count, \
Review Comment:
Fix: wait for 5000ms(default `group.streams.heartbeat.interval.ms` for this test), then check the broker again to see if it has re-solicited.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20817: Add more ducktape system tests scenarios for topology description plugin [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky merged PR #23138:
URL: https://github.com/apache/kafka/pull/23138
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20817: Add more ducktape system tests scenarios for topology description plugin [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #23138:
URL: https://github.com/apache/kafka/pull/23138#discussion_r3814399246
##########
tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py:
##########
@@ -161,3 +163,143 @@ def test_topology_description_not_stored_without_plugin(self, metadata_quorum):
assert int(next(pushed).strip()) == 0, \
"Client logged a successful push despite no plugin being configured on the broker"
processor.stop()
+
+ @cluster(num_nodes=2)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_not_resolicited_after_client_restart(self, metadata_quorum):
+ """
+ Test the situation when the client restarts after already having pushed its
+ topology description successfully. The broker still has storedDescriptionTopologyEpoch
+ matching currentTopologyEpoch, so it must not solicit a second push.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ processor = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.start()
+ monitor.wait_until(self.PUSH_SUCCESS_LOG,
+ timeout_sec=120,
+ err_msg="Streams client did not log a successful topology description push")
+
+ broker_node = self.kafka.nodes[0]
+ solicited_before = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ solicited_before_count = int(next(solicited_before).strip())
+ assert solicited_before_count > 0, \
+ "Broker never solicited the initial topology push despite the plugin being configured"
+
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.restart()
+ monitor.wait_until(self.STREAMS_RUNNING_LOG,
+ timeout_sec=60,
+ err_msg="Never saw 'REBALANCING -> RUNNING' message after client restart " + str(processor.node.account))
+
+ solicited_after = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ assert int(next(solicited_after).strip()) == solicited_before_count, \
+ "Broker re-solicited a topology push after a client restart despite an already-stored, matching-epoch description"
+
+ pushed = processor.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor.LOG_FILE),
+ allow_fail=False)
+ assert int(next(pushed).strip()) == 1, \
+ "Client pushed a topology description again after restart despite the broker not soliciting"
+ processor.stop()
+
+ @cluster(num_nodes=3)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_only_one_member_pushes(self, metadata_quorum):
+ """
+ Test the situation when two members of the same streams group start up together.
+ StreamsGroupTopologyDescriptionManager.armIfNotActive must prevent every member
+ from pushing the same description; only one member's push should succeed,
+ regardless of which member wins the race.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ processor1 = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ processor2 = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ processor1.start()
+ processor2.start()
+
+ def total_push_successes():
+ pushed1 = processor1.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor1.LOG_FILE),
+ allow_fail=False)
+ pushed2 = processor2.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor2.LOG_FILE),
+ allow_fail=False)
+ return int(next(pushed1).strip()) + int(next(pushed2).strip())
+
+ def total_push_failures():
+ failed1 = processor1.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_FAILED_LOG, processor1.LOG_FILE),
+ allow_fail=False)
+ failed2 = processor2.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_FAILED_LOG, processor2.LOG_FILE),
+ allow_fail=False)
+ return int(next(failed1).strip()) + int(next(failed2).strip())
+
+ wait_until(lambda: total_push_successes() >= 1,
+ timeout_sec=120,
+ err_msg=lambda: "Neither streams client logged a successful topology description push"
+ + (" (a non-retriable push failure was logged instead, see client logs)"
+ if total_push_failures() > 0 else ""))
+ assert total_push_failures() == 0, \
+ "A member logged a non-retriable push failure despite a push having succeeded"
+ assert total_push_successes() == 1, \
+ "Expected exactly one member to push the topology description successfully"
+
+ sent1 = processor1.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SENDING_LOG, processor1.LOG_FILE),
+ allow_fail=False)
+ sent2 = processor2.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SENDING_LOG, processor2.LOG_FILE),
+ allow_fail=False)
+ total_sent = int(next(sent1).strip()) + int(next(sent2).strip())
+ assert total_sent == 1, \
+ "Expected exactly one member to send a topology description, got %d" % total_sent
+
+ processor1.stop()
+ processor2.stop()
+
+ @cluster(num_nodes=3)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_resolicited_after_group_delete_and_recreate(self, metadata_quorum):
+ """
+ Test the situation when a streams group is deleted after a successful push, then a
+ new client joins under the same application.id. GroupCoordinatorShard.
+ finalizeStoredDescriptionTopologyEpochAfterDelete clears the deleted group's stored
+ epoch and back-off state, so the new incarnation must be freshly solicited rather
+ than inheriting the "already stored" state left behind by the deleted group.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ group_id = "kafka-streams-system-test-topology-description-plugin"
+ processor = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.start()
+ monitor.wait_until(self.PUSH_SUCCESS_LOG,
+ timeout_sec=120,
+ err_msg="Streams client did not log a successful topology description push")
+
+ processor.stop()
+
+ def group_deleted():
+ return "was successful" in self.kafka.delete_streams_group(group_id)
+
+ wait_until(group_deleted, timeout_sec=30, backoff_sec=2,
Review Comment:
`timeout_sec=30` is tight: each retry shells out a fresh JVM (several seconds), so only a few attempts fit, and if the group empties only after the 10s session timeout under CI load this can spuriously time out. Bump to ~60s.
##########
tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py:
##########
@@ -161,3 +163,143 @@ def test_topology_description_not_stored_without_plugin(self, metadata_quorum):
assert int(next(pushed).strip()) == 0, \
"Client logged a successful push despite no plugin being configured on the broker"
processor.stop()
+
+ @cluster(num_nodes=2)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_not_resolicited_after_client_restart(self, metadata_quorum):
+ """
+ Test the situation when the client restarts after already having pushed its
+ topology description successfully. The broker still has storedDescriptionTopologyEpoch
+ matching currentTopologyEpoch, so it must not solicit a second push.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ processor = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.start()
+ monitor.wait_until(self.PUSH_SUCCESS_LOG,
+ timeout_sec=120,
+ err_msg="Streams client did not log a successful topology description push")
+
+ broker_node = self.kafka.nodes[0]
+ solicited_before = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ solicited_before_count = int(next(solicited_before).strip())
+ assert solicited_before_count > 0, \
+ "Broker never solicited the initial topology push despite the plugin being configured"
+
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.restart()
+ monitor.wait_until(self.STREAMS_RUNNING_LOG,
+ timeout_sec=60,
+ err_msg="Never saw 'REBALANCING -> RUNNING' message after client restart " + str(processor.node.account))
+
+ solicited_after = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ assert int(next(solicited_after).strip()) == solicited_before_count, \
+ "Broker re-solicited a topology push after a client restart despite an already-stored, matching-epoch description"
+
+ pushed = processor.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor.LOG_FILE),
+ allow_fail=False)
+ assert int(next(pushed).strip()) == 1, \
+ "Client pushed a topology description again after restart despite the broker not soliciting"
+ processor.stop()
+
+ @cluster(num_nodes=3)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_only_one_member_pushes(self, metadata_quorum):
+ """
+ Test the situation when two members of the same streams group start up together.
+ StreamsGroupTopologyDescriptionManager.armIfNotActive must prevent every member
+ from pushing the same description; only one member's push should succeed,
+ regardless of which member wins the race.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ processor1 = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ processor2 = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ processor1.start()
+ processor2.start()
+
+ def total_push_successes():
+ pushed1 = processor1.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor1.LOG_FILE),
+ allow_fail=False)
+ pushed2 = processor2.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor2.LOG_FILE),
+ allow_fail=False)
+ return int(next(pushed1).strip()) + int(next(pushed2).strip())
+
+ def total_push_failures():
+ failed1 = processor1.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_FAILED_LOG, processor1.LOG_FILE),
+ allow_fail=False)
+ failed2 = processor2.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_FAILED_LOG, processor2.LOG_FILE),
+ allow_fail=False)
+ return int(next(failed1).strip()) + int(next(failed2).strip())
+
+ wait_until(lambda: total_push_successes() >= 1,
Review Comment:
This reads the counts as soon as the *first* success appears, but `start()` is sequential so processor2 may not have joined yet — a regression where it also pushes could land right after these asserts and be missed. Wait for both members to reach RUNNING (plus a heartbeat interval) before counting; that also actually exercises the concurrent-arm race the docstring describes, which the head start otherwise skips.
##########
tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py:
##########
@@ -161,3 +163,143 @@ def test_topology_description_not_stored_without_plugin(self, metadata_quorum):
assert int(next(pushed).strip()) == 0, \
"Client logged a successful push despite no plugin being configured on the broker"
processor.stop()
+
+ @cluster(num_nodes=2)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_not_resolicited_after_client_restart(self, metadata_quorum):
+ """
+ Test the situation when the client restarts after already having pushed its
+ topology description successfully. The broker still has storedDescriptionTopologyEpoch
+ matching currentTopologyEpoch, so it must not solicit a second push.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ processor = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.start()
+ monitor.wait_until(self.PUSH_SUCCESS_LOG,
+ timeout_sec=120,
+ err_msg="Streams client did not log a successful topology description push")
+
+ broker_node = self.kafka.nodes[0]
+ solicited_before = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ solicited_before_count = int(next(solicited_before).strip())
+ assert solicited_before_count > 0, \
+ "Broker never solicited the initial topology push despite the plugin being configured"
+
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.restart()
+ monitor.wait_until(self.STREAMS_RUNNING_LOG,
+ timeout_sec=60,
+ err_msg="Never saw 'REBALANCING -> RUNNING' message after client restart " + str(processor.node.account))
+
+ solicited_after = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ assert int(next(solicited_after).strip()) == solicited_before_count, \
+ "Broker re-solicited a topology push after a client restart despite an already-stored, matching-epoch description"
+
+ pushed = processor.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor.LOG_FILE),
+ allow_fail=False)
+ assert int(next(pushed).strip()) == 1, \
+ "Client pushed a topology description again after restart despite the broker not soliciting"
+ processor.stop()
+
+ @cluster(num_nodes=3)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_only_one_member_pushes(self, metadata_quorum):
+ """
+ Test the situation when two members of the same streams group start up together.
+ StreamsGroupTopologyDescriptionManager.armIfNotActive must prevent every member
+ from pushing the same description; only one member's push should succeed,
+ regardless of which member wins the race.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ processor1 = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ processor2 = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ processor1.start()
+ processor2.start()
+
+ def total_push_successes():
+ pushed1 = processor1.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor1.LOG_FILE),
+ allow_fail=False)
+ pushed2 = processor2.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor2.LOG_FILE),
+ allow_fail=False)
+ return int(next(pushed1).strip()) + int(next(pushed2).strip())
+
+ def total_push_failures():
+ failed1 = processor1.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_FAILED_LOG, processor1.LOG_FILE),
+ allow_fail=False)
+ failed2 = processor2.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_FAILED_LOG, processor2.LOG_FILE),
+ allow_fail=False)
+ return int(next(failed1).strip()) + int(next(failed2).strip())
+
+ wait_until(lambda: total_push_successes() >= 1,
+ timeout_sec=120,
+ err_msg=lambda: "Neither streams client logged a successful topology description push"
+ + (" (a non-retriable push failure was logged instead, see client logs)"
+ if total_push_failures() > 0 else ""))
+ assert total_push_failures() == 0, \
Review Comment:
`total_push_failures() == 0` doesn't guard the "both members pushed" case. The broker accepts a second push at the same epoch (`validateStreamsGroupTopologyDescriptionUpdate` only checks member + epoch match), so a regression where both members push shows up as two *successes*, not a failure. `successes == 1` and `sent == 1` are the real guards — consider dropping this line, or fix the docstring/`err_msg`, which imply a losing member would log a failure.
##########
tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py:
##########
@@ -161,3 +163,143 @@ def test_topology_description_not_stored_without_plugin(self, metadata_quorum):
assert int(next(pushed).strip()) == 0, \
"Client logged a successful push despite no plugin being configured on the broker"
processor.stop()
+
+ @cluster(num_nodes=2)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_not_resolicited_after_client_restart(self, metadata_quorum):
+ """
+ Test the situation when the client restarts after already having pushed its
+ topology description successfully. The broker still has storedDescriptionTopologyEpoch
+ matching currentTopologyEpoch, so it must not solicit a second push.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ processor = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.start()
+ monitor.wait_until(self.PUSH_SUCCESS_LOG,
+ timeout_sec=120,
+ err_msg="Streams client did not log a successful topology description push")
+
+ broker_node = self.kafka.nodes[0]
+ solicited_before = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ solicited_before_count = int(next(solicited_before).strip())
+ assert solicited_before_count > 0, \
+ "Broker never solicited the initial topology push despite the plugin being configured"
+
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.restart()
+ monitor.wait_until(self.STREAMS_RUNNING_LOG,
+ timeout_sec=60,
+ err_msg="Never saw 'REBALANCING -> RUNNING' message after client restart " + str(processor.node.account))
+
+ solicited_after = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ assert int(next(solicited_after).strip()) == solicited_before_count, \
+ "Broker re-solicited a topology push after a client restart despite an already-stored, matching-epoch description"
+
+ pushed = processor.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor.LOG_FILE),
+ allow_fail=False)
+ assert int(next(pushed).strip()) == 1, \
+ "Client pushed a topology description again after restart despite the broker not soliciting"
+ processor.stop()
+
+ @cluster(num_nodes=3)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_only_one_member_pushes(self, metadata_quorum):
+ """
+ Test the situation when two members of the same streams group start up together.
+ StreamsGroupTopologyDescriptionManager.armIfNotActive must prevent every member
Review Comment:
`armIfNotActive` is defined on `StreamsGroupTopologyDescriptionBackoff`; the Manager only calls it. Minor, but the class name here is off.
##########
tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py:
##########
@@ -161,3 +163,143 @@ def test_topology_description_not_stored_without_plugin(self, metadata_quorum):
assert int(next(pushed).strip()) == 0, \
"Client logged a successful push despite no plugin being configured on the broker"
processor.stop()
+
+ @cluster(num_nodes=2)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_not_resolicited_after_client_restart(self, metadata_quorum):
+ """
+ Test the situation when the client restarts after already having pushed its
+ topology description successfully. The broker still has storedDescriptionTopologyEpoch
+ matching currentTopologyEpoch, so it must not solicit a second push.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ processor = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.start()
+ monitor.wait_until(self.PUSH_SUCCESS_LOG,
+ timeout_sec=120,
+ err_msg="Streams client did not log a successful topology description push")
+
+ broker_node = self.kafka.nodes[0]
+ solicited_before = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ solicited_before_count = int(next(solicited_before).strip())
+ assert solicited_before_count > 0, \
+ "Broker never solicited the initial topology push despite the plugin being configured"
+
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.restart()
+ monitor.wait_until(self.STREAMS_RUNNING_LOG,
+ timeout_sec=60,
+ err_msg="Never saw 'REBALANCING -> RUNNING' message after client restart " + str(processor.node.account))
+
+ solicited_after = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ assert int(next(solicited_after).strip()) == solicited_before_count, \
Review Comment:
This captures the broker solicit count once, right at RUNNING, so a re-solicit arriving on a later heartbeat would be missed. Re-checking after a short settle would make the "no re-solicit" guard robust to a delayed regression.
##########
tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py:
##########
@@ -161,3 +163,143 @@ def test_topology_description_not_stored_without_plugin(self, metadata_quorum):
assert int(next(pushed).strip()) == 0, \
"Client logged a successful push despite no plugin being configured on the broker"
processor.stop()
+
+ @cluster(num_nodes=2)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_not_resolicited_after_client_restart(self, metadata_quorum):
+ """
+ Test the situation when the client restarts after already having pushed its
+ topology description successfully. The broker still has storedDescriptionTopologyEpoch
+ matching currentTopologyEpoch, so it must not solicit a second push.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ processor = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.start()
+ monitor.wait_until(self.PUSH_SUCCESS_LOG,
+ timeout_sec=120,
+ err_msg="Streams client did not log a successful topology description push")
+
+ broker_node = self.kafka.nodes[0]
+ solicited_before = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ solicited_before_count = int(next(solicited_before).strip())
+ assert solicited_before_count > 0, \
+ "Broker never solicited the initial topology push despite the plugin being configured"
+
+ with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+ processor.restart()
+ monitor.wait_until(self.STREAMS_RUNNING_LOG,
+ timeout_sec=60,
+ err_msg="Never saw 'REBALANCING -> RUNNING' message after client restart " + str(processor.node.account))
+
+ solicited_after = broker_node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE),
+ allow_fail=False)
+ assert int(next(solicited_after).strip()) == solicited_before_count, \
+ "Broker re-solicited a topology push after a client restart despite an already-stored, matching-epoch description"
+
+ pushed = processor.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor.LOG_FILE),
+ allow_fail=False)
+ assert int(next(pushed).strip()) == 1, \
+ "Client pushed a topology description again after restart despite the broker not soliciting"
+ processor.stop()
+
+ @cluster(num_nodes=3)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_only_one_member_pushes(self, metadata_quorum):
+ """
+ Test the situation when two members of the same streams group start up together.
+ StreamsGroupTopologyDescriptionManager.armIfNotActive must prevent every member
+ from pushing the same description; only one member's push should succeed,
+ regardless of which member wins the race.
+ """
+ self.setup_kafka(plugin_enabled=True)
+ processor1 = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ processor2 = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+ processor1.start()
+ processor2.start()
+
+ def total_push_successes():
+ pushed1 = processor1.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor1.LOG_FILE),
+ allow_fail=False)
+ pushed2 = processor2.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor2.LOG_FILE),
+ allow_fail=False)
+ return int(next(pushed1).strip()) + int(next(pushed2).strip())
+
+ def total_push_failures():
+ failed1 = processor1.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_FAILED_LOG, processor1.LOG_FILE),
+ allow_fail=False)
+ failed2 = processor2.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_FAILED_LOG, processor2.LOG_FILE),
+ allow_fail=False)
+ return int(next(failed1).strip()) + int(next(failed2).strip())
+
+ wait_until(lambda: total_push_successes() >= 1,
+ timeout_sec=120,
+ err_msg=lambda: "Neither streams client logged a successful topology description push"
+ + (" (a non-retriable push failure was logged instead, see client logs)"
+ if total_push_failures() > 0 else ""))
+ assert total_push_failures() == 0, \
+ "A member logged a non-retriable push failure despite a push having succeeded"
+ assert total_push_successes() == 1, \
+ "Expected exactly one member to push the topology description successfully"
+
+ sent1 = processor1.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SENDING_LOG, processor1.LOG_FILE),
+ allow_fail=False)
+ sent2 = processor2.node.account.ssh_capture(
+ "grep -c '%s' %s || true" % (self.PUSH_SENDING_LOG, processor2.LOG_FILE),
+ allow_fail=False)
+ total_sent = int(next(sent1).strip()) + int(next(sent2).strip())
+ assert total_sent == 1, \
+ "Expected exactly one member to send a topology description, got %d" % total_sent
+
+ processor1.stop()
+ processor2.stop()
+
+ @cluster(num_nodes=3)
+ @matrix(metadata_quorum=[quorum.combined_kraft])
+ def test_topology_description_resolicited_after_group_delete_and_recreate(self, metadata_quorum):
+ """
+ Test the situation when a streams group is deleted after a successful push, then a
+ new client joins under the same application.id. GroupCoordinatorShard.
+ finalizeStoredDescriptionTopologyEpochAfterDelete clears the deleted group's stored
Review Comment:
Minor doc nit: on `GroupCoordinatorShard` this method is `finalizeStoredDescriptionTopologyEpochAfterDeleteBatch`; the non-batch `finalizeStoredDescriptionTopologyEpochAfterDelete` lives on `GroupMetadataManager`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org