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