KAFKA-20697: Add cross-version system tests for the streams rebalance protocol by suzhiking · Pull Request #23146 · apache/kafka

suzhiking · 2026-08-12T23:29:10Z

Kafka 4.4 bumps StreamsGroupHeartbeat and StreamsGroupDescribe from v0 to v1 (KIP-1331, KIP-1357, KAFKA-20744). The new fields only flow when both sides are 4.4+, so this adds ducktape coverage for mixed-version deployments in both directions, using 4.2.1/4.3.1 as the older side (4.2 is the earliest release with streams group support). The tests drive the StreamsUpgradeTest harness with group.protocol=streams.

New streams_protocol_cross_version_test.py (7 tests, 15 matrix cases):

new client / old broker: reaches RUNNING, logs acceptableRecoveryLag=not provided (older broker) exactly when the response was v0, no UnsupportedVersionException
old client / new broker: reaches RUNNING, the ignorable v1 lag field is dropped cleanly at v0
MISSING_CLIENT_TAGS is only sent to v1 clients, and not sent when the required tag is configured
kafka-streams-groups.sh --describe works cross-version in both directions; --describe --topology against an old broker fails diagnosably rather than hanging (graceful degradation is a possible follow-up; the assertion is deliberately loose so it survives one)
a plugin-configured new broker reports no stored topology description for an old client

Extended streams_topology_description_plugin_test.py (+1 test): a new client is never solicited for a topology push by an old broker, since TopologyDescriptionRequired only exists in response v1. setup_kafka gained an optional broker_version parameter; existing tests unchanged.

The warm-up scenario from the ticket (old clients must never get warm-up tasks) is intentionally left out: AssignmentRefiner is still a stub, so the test would pass vacuously until KAFKA-20665 lands.

Reviewers: Matthias J. Sax matthias@confluent.io

… protocol Kafka 4.4 bumps the streams rebalance protocol RPCs, StreamsGroupHeartbeat (apiKey 88) and StreamsGroupDescribe (apiKey 89), from version 0 to version 1. Heartbeat response v1 replaces the v0 int32 AcceptableRecoveryLag field with an ignorable int64 AcceptableRecoveryLag, adds TopologyDescriptionRequired (KIP-1331), and permits the new MISSING_CLIENT_TAGS status code (KAFKA-20744), which the coordinator only returns on v1 requests since v0 clients do not know the code. Describe v1 adds IncludeTopologyDescription on the request and TopologyDescription, TopologyDescriptionStatus and AssignorName on the response (KIP-1331, KIP-1357). New fields flow only when both sides are 4.4+, so mixed-version deployments must keep working in both directions. Add ducktape coverage using 4.2.1/4.3.1 as the older side, 4.2 being the earliest release that supports streams groups (streams.version 1 requires metadata version 4.2-IV1). The tests drive the StreamsUpgradeTest harness, which ships in the 4.2/4.3 streams test jars, with group.protocol=streams. New streams_protocol_cross_version_test.py: - test_new_client_old_broker: a trunk client against 4.2/4.3/dev brokers reaches RUNNING; the client logs acceptableRecoveryLag as not provided exactly when the broker answered with a v0 heartbeat response. - test_old_client_new_broker: a 4.2/4.3 client against a trunk broker reaches RUNNING; the ignorable v1 lag field is dropped cleanly at v0. - test_missing_client_tags_status_gated_by_rpc_version: with rack-aware assignment tags required but not configured on the client, only a trunk client is sent the MISSING_CLIENT_TAGS status; 4.2/4.3 clients never see it. - test_missing_client_tags_status_absent_when_tag_configured: a trunk client that does configure the required tag receives no such status, proving the version gate is not simply never firing. - test_describe_new_tool_old_broker: plain kafka-streams-groups.sh --describe from trunk works against 4.2/4.3 brokers; --describe --topology cannot be served at describe v0 and must fail diagnosably rather than hang. - test_describe_old_tool_new_broker: the 4.2/4.3 CLI describes a group hosted on a trunk broker. - test_topology_description_not_stored_for_old_client: a plugin-configured trunk broker reports no stored topology description for a 4.2/4.3 client, which can never be asked to push one. Extended streams_topology_description_plugin_test.py: - test_no_push_solicited_by_old_broker: a trunk client against a 4.2/4.3 broker is never solicited for a topology description push and never sends one, since TopologyDescriptionRequired only exists in response v1. The headline scenario of KAFKA-20697, verifying that clients which cannot report task offsets are never assigned warm-up tasks, is deliberately not covered yet: the group coordinator's AssignmentRefiner is still a stub that returns the target assignment unchanged, so such a test would pass vacuously until KAFKA-20665 lands.

mjsax · 2026-08-20T23:42:53Z

+                                 server_prop_overrides=server_prop_overrides)
+        self.kafka.set_version(KafkaVersion(broker_version))
+        self.kafka.start()
+        self.kafka.run_features_command("upgrade", "streams.version", 1)


Why is this necessary? Should be set to 1 by default.

Guess this might be legacy code from 4.1 EA release, which just got c&p around. We might want to do some cleanup (maybe new PR -- if small enough, also ok to just add to this PR) to remove from other tests, too

Claude also pointed out, that there is some other potential legacy code. use_streams_groups also sets UNSTABLE_API_VERSIONS_ENABLE and UNSTABLE_FEATURE_VERSIONS_ENABLE in kafka.py -- this also sounds like 4.1 EA stuff, which required to enable unable API/Features to allows us to enable "streams" protocol.

Sure, I'll remove them in this PR. And yes, I copied it from streams_topology_description_plugin_test.py

mjsax · 2026-08-20T23:46:37Z

+    @cluster(num_nodes=2)
+    @matrix(broker_version=[str(LATEST_4_2), str(LATEST_4_3), str(DEV_BRANCH)],
+            metadata_quorum=[quorum.combined_kraft])
+    def test_new_client_old_broker(self, broker_version, metadata_quorum):


If we say "new client old broker" it seems broker_version above should only be 4.2/4.2 but not DEV_BRANCH?

But the test below does assert that acceptable-recovery-lag is provided when broker is new -- so maybe the test name is just wrong? -- On the other hand, for 4.4/4.4 broker/client we can cover this in unit/integration tests, so it seems not necessary to add as a system test?

correct, I'll remove dev branch from test

mjsax · 2026-08-20T23:50:10Z

+
+        lag_not_provided = self.count_in_file(processor.node, self.OLD_BROKER_LAG_LOG, processor.LOG_FILE)
+        if broker_version == str(DEV_BRANCH):
+            assert lag_not_provided == 0, \


Is this check sufficient? We only have a negative test, but not a positive test -- for a new broker, should we assert that we see that acceptable.recover.lag=10000 (I believe 10K is the default) gets logged?

will remove this branch

mjsax · 2026-08-20T23:55:31Z

+        assert self.count_unsupported_version_errors(processor.node, processor.LOG_FILE) == 0, \
+            "The 4.4 client hit an UnsupportedVersionException against a %s broker" % broker_version
+
+        processor.stop()


Should we also assert that there was no error on the broker? Or would this be redundant? Not sure from top of my head, but if the broker received something unexpected, it would also log an error.

I am particularly thinking about task-offset-sum / task-offset-end-sum in the HB request-- these fields are already present in v0, but should never be set (and I believe 4.2/4.3 brokers check that both fields are always zero, and we did lift this check with 4.4 brokers). Worth checking if we made similar changes for other field.

Also wondering about now populated fields in the response like taskOffsetIntervalMs ?

agree, gonna add test for taskOffsetIntervalMs

mjsax · 2026-08-21T01:28:26Z

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


Do we need this test as system test? Seems this can be covered (and hopefully is already covered) as integration test?

mjsax · 2026-08-21T01:32:28Z

+    def test_describe_new_tool_old_broker(self, broker_version, metadata_quorum):
+        """
+        The 4.4 kafka-streams-groups.sh must describe a group hosted on a 4.2/4.3 broker. Plain
+        --describe stays within describe v0 and must succeed.


What about "assignor name" -- older brokers does not support it, so we need a negative check (and check that we don't crash?)

mjsax · 2026-08-21T01:40:46Z

+            metadata_quorum=[quorum.combined_kraft])
+    def test_no_push_solicited_by_old_broker(self, broker_version, metadata_quorum):
+        """
+        Test a 4.4 client against a 4.2/4.3 broker. The broker solicits a topology description push


Test a 4.4 client -- this comment is only "correct" on 4.4 branch, but on trunk it would already be 4.5.

Can we phrase this differently, so the comment does not outdate, each time trunk version is bumped. Might apply to other comments, too -- just noticed here for the first time.

mjsax · 2026-08-21T01:42:17Z

+        Test a 4.4 client against a 4.2/4.3 broker. The broker solicits a topology description push
+        by setting topologyDescriptionRequired on the heartbeat response, but that field only exists
+        in response version 1 (KIP-1331), so a 4.2/4.3 broker can never ask. Such a broker also does
+        not know the plugin config and only warns about it, which this configures deliberately: even


Such a broker also does not know the plugin config and only warns about it

Not sure what this means? Are you saying, if the config is provided broker side the broker just logs "unknown" -- sure, but this seems to be irrelevant for this test?

even with the config present, the older broker must not cause the newer client to push.

Similar. An older broker really cannot "cause" a new client to push -- and the test is not about this. It's really about a buggy client which would push even if there was no signal from the broker, that the broker understand the RPC.

will remove

github-actions Bot added triage PRs from the community tests Test fixes (including flaky tests) labels Aug 12, 2026

suzhiking force-pushed the KAFKA-20697-cross-version-system-tests branch from 2534f7a to 9d12673 Compare August 13, 2026 00:05

mjsax added ci-approved and removed triage PRs from the community labels Aug 14, 2026

mjsax reviewed Aug 21, 2026

View reviewed changes

Uh oh!

Conversation

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

Uh oh!

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!

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!

2 participants

suzhiking commented Aug 12, 2026 •

edited by github-actions Bot

Loading