KAFKA-20697: Add cross-version system tests for the streams rebalance protocol

[suzhiking]

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 because v0 clients do not know the code. The heartbeat request is byte-identical between v0 and v1; it was bumped so the response version is negotiated.
• 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. This PR adds ducktape coverage for that, using 4.2.1/4.3.1 as the older side — 4.2 is 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 and accepts arbitrary config from its properties file, with group.protocol=streams.

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

Test	Matrix	Asserts
test_new_client_old_broker	broker 4.2.1/4.3.1/dev	Client reaches RUNNING; logs acceptableRecoveryLag=not provided (older broker) exactly when the broker answered v0; no UnsupportedVersionException
test_old_client_new_broker	client 4.2.1/4.3.1	Old client reaches RUNNING against a dev broker; the ignorable v1 lag field is dropped cleanly at v0
test_missing_client_tags_status_gated_by_rpc_version	client 4.2.1/4.3.1/dev	With group.streams.rack.aware.assignment.tags set and no client tag configured, only the dev client is sent MISSING_CLIENT_TAGS; 4.2/4.3 clients never see it
test_missing_client_tags_status_absent_when_tag_configured	—	A dev client that does configure the required tag receives no such status (proves the version gate is not simply never firing)
test_describe_new_tool_old_broker	broker 4.2.1/4.3.1	Plain kafka-streams-groups.sh --describe from dev works against old brokers; --describe --topology cannot be served at describe v0 and must fail diagnosably rather than hang
test_describe_old_tool_new_broker	tool 4.2.1/4.3.1	The old CLI describes a group hosted on a dev broker
test_topology_description_not_stored_for_old_client	client 4.2.1/4.3.1	A plugin-configured dev broker reports no stored topology description for an old client, which can never be asked to push one

Extended: streams_topology_description_plugin_test.py (+1 test, 2 matrix cases)

test_no_push_solicited_by_old_broker (broker 4.2.1/4.3.1): a dev client against an old broker is never solicited for a topology description push and never sends one, since TopologyDescriptionRequired only exists in response v1. setup_kafka gained an optional broker_version parameter; the three existing tests are unchanged.

Deliberately not covered yet

The headline scenario of KAFKA-20697 — verifying that clients which cannot report task offsets are never assigned warm-up tasks — is not testable on current trunk: AssignmentRefiner.refine() is still a stub that returns the target assignment unchanged, so the assertion would pass vacuously. That test should follow once KAFKA-20665 lands.

Notes for reviewers

• test_describe_new_tool_old_broker asserts only that the --topology failure names the unsupported field or version. Today the CLI exits 1 with a generic message and a stack trace (the UnsupportedVersionException from serializing the non-ignorable IncludeTopologyDescription at v0 falls through to the top-level catch (Throwable)); the --delete-offsets path already models a friendlier "not supported by the broker version" message. Whether describe should degrade gracefully is left as a follow-up; the assertion is deliberately loose so it survives such a fix.
• End-to-end record flow is intentionally not asserted: reaching RUNNING already requires a successful join, topology initialization, an assignment from the coordinator, and task creation, which is the protocol surface under test here.

Testing

Both files byte-compile, and an AST-level check validates that every @matrix key matches its test method signature (the checker reproduces the known 30 cases of streams_broker_compatibility_test.py). All asserted log lines and CLI outputs were verified against the shipped sources of 4.2.1, 4.3.1 and trunk (e.g. the State transition from REBALANCING to RUNNING line exists in all three; the MISSING_CLIENT_TAGS detail string and the not provided (older broker) rendering are trunk-side). A ducktape run of the two suites has not been executed yet; results will be posted on this PR.

[suzhiking]

Superseded by a re-opened PR from a renamed branch.