KAFKA-20629: Add ducktape system tests for topology description plugin by lucliu1108 · Pull Request #22769 · apache/kafka

lucliu1108 · 2026-07-06T19:00:21Z

Summary

Introduces the system-test infrastructure for KIP-1331 (Streams Group
Topology Description Plugin):

New Java driver TopologyDescriptionPluginSystemTest that starts a
small streams app and exits on signal.
New ducktape suite streams_topology_description_plugin_test with 3
scenarios:
1. push round-trip succeeds when the plugin is configured and the
  client default push is on;
2. client opt-out via topology.description.push.enabled=false
  prevents the client from ever sending a description;
3. no topology description plugin is configured on the broker.

Add
group.streams.topology.description.plugin.class=org.apache.kafka.server.streams.InMemoryTopologyDescriptionPlugin
to BaseStreamsTest and StreamsBrokerBounceTest (when
use_streams_groups=True). The plugin jar is already on the broker
classpath via the server module.
Enable the InMemory plugin in the streams_broker_compatibility_test
for DEV_VERSION.

Reviewers: Alieh Saeedi asaeedi@confluent.io, TengYao Chi
frankvicky@apache.org

…n (1/2) Introduces the system-test infrastructure for KIP-1331 (Streams Group Topology Description Plugin): - New Java driver TopologyDescriptionPluginSystemTest that starts a small streams app and exits on signal. - New ducktape suite streams_topology_description_plugin_test with three scenarios: * push round-trip succeeds when the plugin is configured and the client default push is on; * client opt-out via topology.description.push.enabled=false prevents the client from ever sending a description; * no plugin on the broker means no solicitation is issued. - KafkaService constructor parameter streams_group_topology_description_plugin_class and matching config_property constant to propagate group.streams.topology.description.plugin.class to brokers. - StreamsTopologyDescriptionPluginService and the supporting TOPOLOGY_DESCRIPTION_PUSH_ENABLED property constant for the new driver. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

…ystem tests (2/2) Wires the InMemoryTopologyDescriptionPlugin into the streams tests that already exercise the streams group protocol, plus a smoke entry in the broker compatibility matrix: - BaseStreamsTest passes the plugin to KafkaService; the plugin is on for every extender, since BaseStreamsTest always uses use_streams_groups=True. - StreamsBrokerBounceTest passes the plugin only when group_protocol='streams', mirroring how use_streams_groups is set. - StreamsBrokerCompatibilityTest adds the plugin class as a raw server_prop_overrides row and appends str(DEV_VERSION) to both broker_version matrices. On the DEV broker, the plugin is loaded at startup; older brokers ignore the unknown config key. The classic-protocol driver does not exercise the push path, so this is a regression smoke that the plugin doesn't break broker startup. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

aliehsaeedii

Thanks @lucliu1108 for the PR. I made the first round:)

aliehsaeedii · 2026-07-07T10:54:09Z

+        processor = StreamsTopologyDescriptionPluginService(
+            self.test_context, self.kafka, topology_description_push_enabled=False)
+        processor.start()
+        time.sleep(30)


The two negative tests can pass vacuously: processor.start() only waits for STREAMS-STARTED, which is printed right after KafkaStreams#start() returns — before the client has joined the streams group. If the app crashes or never manages to join during the 30s sleep, the grep -c == 0 assertions still pass even though nothing was actually verified.

Consider waiting for a positive liveness signal before doing the negative grep — e.g. wait for the State transition from REBALANCING to RUNNING line in the streams log (which implies successful heartbeats with the coordinator), and only then assert the push logs are absent. Same applies to test_topology_description_not_stored_without_plugin below.

Fix: added assertions to wait for client to join the group.

aliehsaeedii · 2026-07-07T10:54:09Z

+        """
+        Test the situation when the broker has the topology description plugin configured
+        but the client opts out via topology.description.push.enabled=false. The client
+        should never send a topology description even though the broker solicits one.


Nit on the docstring: "even though the broker solicits one" is true broker-side, but it isn't observable in this test. When the client opts out, StreamThread never sets the wire topology description, so the Broker requested topology description push log in StreamsGroupHeartbeatRequestManager is suppressed too (the condition requires a non-null wire description). Maybe reword so a future reader doesn't try to assert PUSH_REQUESTED_LOG > 0 here and get confused when it's 0.

Fix: added additional explanation indicating the "broker requested topology description push" log never appears; an additional assertion on this is also added.

aliehsaeedii · 2026-07-07T10:54:09Z

                                      ["transaction.state.log.replication.factor", "1"],
-                                      ["transaction.state.log.min.isr", "1"]
+                                      ["transaction.state.log.min.isr", "1"],
+                                      ["group.streams.topology.description.plugin.class", INMEMORY_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS]


The PR description says the plugin is enabled "for DEV_VERSION", but this override applies to every broker version in the matrix. It happens to work because group.streams.topology.description.plugin.class only exists on trunk (not in 4.3 or earlier), so older brokers just log an unknown-config warning and ignore it — but that's subtle enough to deserve an inline comment.

Also, since StreamsBrokerCompatibilityService runs with the classic protocol (no streams groups exist in this test), the plugin is instantiated but never exercised. Could you clarify what this addition is meant to verify — that a DEV broker with the plugin configured doesn't affect classic-protocol streams apps? A short comment stating that intent would help.

Fix: I added comments about this config: 1) it is only in actual effect when broker_version>4.3 2) This test is only using classic group so the plugin is not taking effect.

aliehsaeedii · 2026-07-07T10:54:09Z

+                            str(LATEST_4_2),str(LATEST_4_3),str(DEV_BRANCH)],
            metadata_quorum=[quorum.combined_kraft]
            )
+


Stray whitespace-only line added between the @matrix decorator and the test method — please remove.

aliehsaeedii · 2026-07-07T10:54:09Z

+
+    @cluster(num_nodes=2)
+    @matrix(metadata_quorum=[quorum.combined_kraft])
+    def test_topology_description_available_with_plugin(self, metadata_quorum):


Are follow-up scenarios planned for the plugin lifecycle? Three come to mind that the in-memory plugin makes observable with exactly the infrastructure this PR adds:

Broker bounce → re-solicitation: the InMemoryTopologyDescriptionPlugin loses its state on restart, so after a bounce the broker should solicit again and the client should push a second time (happy-path push → bounce → wait for a second Topology description pushed successfully).

Client restart with the description already stored: restart only the streams app while the broker keeps running — since storedDescriptionTopologyEpoch == currentTopologyEpoch, the broker should not solicit again.

Multiple members, single push: with 2+ clients in the group, the solicitation back-off (StreamsGroupTopologyDescriptionManager.armIfNotActive) should prevent every member from pushing the same description — e.g. assert the second member never sends while the first push succeeds.

No need to grow this PR — just wondering whether these are on the part-2 list.

Thanks for the ideas! I will append more scenarios in a followup PR.

frankvicky · 2026-07-07T13:56:25Z

+        processor.start()
+        time.sleep(30)
+        # count how many lines in streams.log contain Broker requested topology description push
+        solicited = processor.node.account.ssh_capture(


This only checks Broker requested topology description push count == 0. Belt-and-suspenders: also assert PUSH_SENDING_LOG == 0 and PUSH_SUCCESS_LOG == 0. Cheap defense against a client-side regression where the client pushes uninvited.

Fix: added the 2 client-side assertions

StreamsGroupHeartbeatRequestManager.java:657-660：

if (data.topologyDescriptionRequired() && streamsRebalanceData.wireTopologyDescription() != null) { logger.info("Broker requested topology description push");

When topology.description.push.enabled=false, wireTopologyDescription() is null, so this log line is never emitted. That means: regardless of whether the broker sets topologyDescriptionRequired=true, the count will always be 0. This assertion cannot distinguish "broker never solicited" from "broker did solicit but the client silently ignored it" — and the latter is exactly the gap I originally asked to be defended against. To actually verify this, the assertion needs to move to the broker side (for example, "the broker logged that setTopology was requested, but setTopology was never invoked").

Thanks! I added a broker log after the config is solicit on the broker side, and an additional check in the system test for its appearance.

lucliu1108 and others added 9 commits July 1, 2026 12:53

add hb unstable api version

be35f9e

revise comment

897509c

remove unnecessary enable unstable version

de2749f

modify import orders

d88e1ac

revise new dev version

e9e6251

fix style

ecd4ea8

stash part2 change to 1

83289f0

github-actions Bot added triage PRs from the community streams labels Jul 6, 2026

lucliu1108 marked this pull request as ready for review July 6, 2026 19:01

aliehsaeedii reviewed Jul 7, 2026

View reviewed changes

frankvicky added the ci-approved label Jul 7, 2026

frankvicky reviewed Jul 7, 2026

View reviewed changes

lucliu1108 added 3 commits July 7, 2026 13:29

revise test

7a0e8f4

modify error messages

6d24778

add additional broker side log assertion for client-plugin-disabled case

cc5bad7

github-actions Bot removed the triage PRs from the community label Jul 8, 2026

lucliu1108 requested review from aliehsaeedii and frankvicky July 8, 2026 14:47

add broker side log and confirmation for configuration set

d767996

github-actions Bot added the group-coordinator label Jul 9, 2026

frankvicky approved these changes Jul 10, 2026

View reviewed changes

Uh oh!

Conversation

lucliu1108 commented Jul 6, 2026 • edited by github-actions Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Uh oh!

aliehsaeedii left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

lucliu1108 commented Jul 6, 2026 •

edited by github-actions Bot

Loading