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