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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
