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