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