…scription plugin (apache#23138)

## Summary
This PR adds 3 more system test scenarios for streams topology test.
1. `test_topology_description_not_resolicited_after_client_restart`:
client start with description already stored, the broker is already
running -> client restarted and since `storedDescriptionTopologyEpoch ==
currentTopologyEpoch`, the broker does not solicit again.
2. `test_topology_description_only_one_member_pushes`: two members join
the same streams group at the same time and both heartbeat to the broker
-> since `StreamsGroupTopologyDescriptionBackoff.armIfNotActive`'s
check-and-arm on the group's back-off entry is atomic, only one member's
heartbeat gets `TopologyDescriptionRequired=true`; the other member
never sends a push.
3.
`test_topology_description_resolicited_after_group_delete_and_recreate`:
client pushes successfully, then the group is stopped and deleted -> a
new client joins under the same application.id and since the delete
clears the old group's `storedDescriptionTopologyEpoch`, the broker
treats it as a brand-new group and solicits again.

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

(cherry picked from commit c7b0347)

…y description plugin (apache#23413)

The heartbeat- and describe-based topology push protocol added by
KIP-1331 relies on StreamsGroupHeartbeat and StreamsGroupDescribe
negotiating down to earlier versions when talking to a broker or client
that predates the feature, and on the admin client refusing to send a
topology description request a broker cannot understand. This was
previously only checked by unit tests exercising the version-gating
logic directly; nothing ran an actual pre-4.4 Kafka broker or Streams
client against the feature to prove the real wire negotiation degrades
gracefully in both directions.

This adds three system tests covering the missing directions: a Streams
client built from this branch (topology push enabled by default) talking
to a genuine 4.3.1 broker, a genuine 4.3.1 Streams client talking to a
broker on this branch with the plugin configured, and the admin
client/CLI on this branch requesting a topology description from a 4.3.1
broker, which must fail client-side with UnsupportedVersionException
rather than silently describing the group without it.

## Test plan

All three tests were run against real ducker containers (docker-based
ducktape harness) to confirm they exercise the intended code paths
rather than passing vacuously:
- Confirmed via broker/client logs that the pinned old-version processes
actually ran Kafka 4.3.1, not the dev-branch build.
- Confirmed the admin-client test's failure text matches
`UnsupportedVersionException: Attempted to write a non-default
includeTopologyDescription at version 0` with exit code 1.
- Re-ran the full `streams_topology_description_plugin_test.py` suite (9
tests) plus the new smoke test together; all pass.

Reviewers: TengYao Chi <frankvicky@apache.org>
(cherry picked from commit fb48d18)

… description plugin enabled (apache#23498)

Adds a companion-object subclass to JoinGroupRequestTest and
StreamsGroupHeartbeatRequestTest, following the same
@ClusterTestDefaults override pattern already used by
WithAssignmentBatchingDisabledTest, so each suite reruns in full with
the KIP-1331 streams-group topology description plugin
(group.streams.topology.description.plugin.class) configured. This gives
automated proof that the plugin's presence does not change the existing
JoinGroup and Heartbeat semantics these suites already cover.

Testing: ran both parametrized suites locally (`:core:test --tests
"kafka.server.StreamsGroupHeartbeatRequestTest*" --tests
"kafka.server.JoinGroupRequestTest*"`); all cases, including the new
plugin-enabled variants, pass.

Reviewers: TengYao Chi <frankvicky@apache.org>
(cherry picked from commit d81598c)

… protocol (apache#23146)

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>, Lucas Brutschy
<lbrutschy@confluent.io>

(cherry picked from commit 9987c5b)

…d broker (apache#23523)

Adds a system test verifying that a newer (dev/4.4+) Kafka Streams
client using static membership on the `streams` group protocol correctly
detects and fails against an older broker (4.2/4.3) that doesn't support
the `streams` group protocol.

The client sends a `STREAMS_GROUP_HEARTBEAT` request, which the old
broker rejects with `INVALID_REQUEST` ("Static membership is not yet
supported"). The client then transitions to the terminal `ERROR` state
rather than hanging or retrying indefinitely.

Note on scope: this PR originally also covered the reverse case (old
client vs new broker), but that direction fails with a client-side
`ConfigException` during `StreamsConfig` construction, before any broker
contact. This is a local configuration guard rather than a cross-version
compatibility issue, and is already covered by unit tests. Per
discussion on the ticket, that direction has been dropped from this PR.

Testing:

* Verified against Kafka 4.2.1.
* Verified against Kafka 4.3.1.
* Ran the test both in isolation and as part of the full
`StreamsStaticMembershipTest` suite.

Reviewers: Lucas Brutschy <lbrutschy@confluent.io>
(cherry picked from commit 81d676c)

…er tests (apache#23619)

The replayRecords javadoc described it as "the records to replay", even
though it is a boolean flag controlling whether the coordinator applies
records to its in-memory state, not a list of records.

The two streams initial-rebalance-delay tests also could not tell
whether the target assignment they observed came from the delay timer
actually replaying its records, or from a later heartbeat recomputing
the assignment on its own, since both tests only checked the group state
after an additional heartbeat had already run past the timer firing.
They now assert the in-memory target assignment right after the timer
fires and before any further heartbeat, and the first test performs an
extra heartbeat inside the delay window so the topology is configured on
the persisted group rather than a throwaway instance, which is required
for the timer to compute and replay the assignment in the first place.

Reviewers: Sean Quah <squah@confluent.io>
(cherry picked from commit 3c7b066)

…s.ERROR (apache#23657)

The describe --topology command in StreamsGroupCommand already handles
the ERROR topology description status correctly, returning a non-zero
exit code and printing a message pointing to the broker logs. This path
was not covered by a test, unlike the AVAILABLE and NOT_STORED statuses
which already have dedicated tests in StreamsGroupCommandTest.

This PR adds a test for the ERROR status following the same pattern as
the existing coverage, asserting both the exit code and the printed
error message.

Reviewers: Andy Huang (github:andyhuangdev)
(cherry picked from commit d6565d5)

…pdate RPC end-to-end (apache#23658)

This adds an integration test alongside
`TopologyDescriptionPluginIntegrationTest` that runs a Streams app with
`topology.description.push.enabled=false` against a broker configured
with the tracking topology description plugin used in that test class.
Today the client-side suppression is only unit-tested at

`StreamThreadTest.shouldNotPopulateWireTopologyDescriptionWhenPushDisabled`,
which only checks that the client does not populate the wire topology
description object; it does not prove the behavior end-to-end against a
real broker.

The new test keeps the group alive without a stored topology
description, so the broker keeps soliciting the push on every heartbeat
exactly as it does for a push-enabled client (same mechanism already
exercised by `shouldPushTopologyDescriptionToPluginAfterJoin` in the
same class). Waiting across several heartbeat intervals and asserting
the plugin's `setTopology` is never invoked and the group's topology
description status stays `NOT_STORED` shows that the suppression is a
genuine client-side opt-out rather than something that happens to work
because the broker never asks.

### Test plan
- `./gradlew :streams:integration-tests:test --tests

"org.apache.kafka.streams.integration.TopologyDescriptionPluginIntegrationTest"`
passes, including the new
`shouldNotPushTopologyDescriptionWhenDisabledByClient` test.

Reviewers: Bill Bejeck <bbejeck@apache.org>
(cherry picked from commit ed44e4b)

…ailure surfacing (apache#23670)

StreamsGroupTopologyDescriptionRequestTest only ever exercises
InMemoryTopologyDescriptionPlugin, which never fails, so the permanent
and transient failure classification for the
StreamsGroupTopologyDescriptionUpdate RPC, and the DeleteGroups
plugin-failure path, were proven only against mocks in
GroupCoordinatorServiceTopologyDescriptionTest, never against a real
broker/RPC round trip.

This adds FailingTopologyDescriptionPlugin, a configurable test double
whose setTopology and deleteTopology calls can be told to fail on
demand, and three new cluster tests built on it. A permanent setTopology
failure surfaces STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED with the
plugin's message and stops the broker from re-soliciting a push at that
epoch. A transient failure returns the same error code but arms the
heartbeat back-off instead of ratcheting the epoch. A DeleteGroups
plugin failure returns GROUP_DELETION_FAILED without tombstoning the
group, and a retried delete succeeds once the plugin recovers.

Reviewers: TengYao Chi <frankvicky@apache.org>, Andy Huang
 <165141963+andyhuangdev@users.noreply.github.com>
(cherry picked from commit c553733)

…eanup (apache#23747)

Converting an empty streams group to a classic group requires the broker
to delete the group's stored topology description from the plugin first.
This path was only unit-tested against mocks; no test sent a real
classic JoinGroupRequest to a running broker, and the failed-join and
periodic-cleanup halves were never chained together.

This adds three single-broker integration tests to
StreamsGroupTopologyDescriptionRequestTest. The first joins a classic
member to an empty streams group with a stored topology and checks that
the broker calls plugin.deleteTopology and converts the group. The
second makes deleteTopology fail and checks that the join is rejected
with the retriable REBALANCE_IN_PROGRESS while the group stays a streams
group. The third chains the failure with the periodic cleanup cycle:
after the plugin recovers, the cycle clears the stored topology without
any client involvement, and a later classic join converts the group.

FailingTopologyDescriptionPlugin gains a per-group counter of
deleteTopology invocations, so the tests can assert that the broker
actually reached the plugin.

Reviewers: TengYao Chi <frankvicky@apache.org>
(cherry picked from commit 3c1b3f2)

…ache#23705)

Fixes KAFKA-21212

When a Streams static member restarted after removing `client.rack`, the
coordinator retained the previous rack ID. During static member
replacement, the coordinator copies the previous member’s metadata, then
applies the incoming updates. Because `maybeUpdateRackId` treats null as
unchanged, the joining request could not clear the copied rack ID. This
left stale rack information in group descriptions and custom assignor
inputs.

This change distinguishes joining requests from regular heartbeats,
following the existing handling of `userEndpoint`. Joining requests now
apply the rack ID directly, including null to clear the previous value.
Regular heartbeats continue to preserve the existing rack ID when the
field is omitted. Clearing a previously configured rack ID also allows
the existing metadata change detection to bump the group epoch and
trigger assignment recomputation.

Reviewers: Lucas Brutschy <lbrutschy@confluent.io>
(cherry picked from commit 0661cd0)

…reported. (apache#23768)

Fixes KAFKA-21255.

Return the current assignment in StreamsGroupHeartbeat responses
whenever the request includes all three owned task sets(active, standby,
and warmup) including empty sets. This allows clients to recover a
missed assignment after heartbeat response loss, even when processing
the retry does not change the coordinator’s assignment.

Reviewers: Lucas Brutschy <lbrutschy@confluent.io>
(cherry picked from commit 5f684cc)

Backport note: dropped the change to a test of the assignment refiner, which is not on 4.4. In the new tests, added the missing StreamsGroupTestUtil imports and used getDefaultAssignmentConfigs(), since 4.4 stores the last assignment configs as a map.