KAFKA-20627: Add broker integration tests for topology description plugin

[aliehsaeedii]

Adds the integration test coverage for the streams group topology
description plugin (KIP-1331) requested in
KAFKA-20627:

Broker-level RPC tests (core)

• StreamsGroupTopologyDescriptionRequestTest (plugin configured
  with InMemoryTopologyDescriptionPlugin):
  • heartbeat solicits the push via TopologyDescriptionRequired, the
    push → describe lifecycle returns AVAILABLE with the stored topology,
    and there is no re-solicitation once stored
  • validation errors: empty group/member id → INVALID_REQUEST,
    unknown group → GROUP_ID_NOT_FOUND, unknown member →
    UNKNOWN_MEMBER_ID, stale topology epoch → INVALID_REQUEST
  • DeleteGroups succeeds on a group with a stored description and the
    group is gone afterwards
• StreamsGroupTopologyDescriptionNoPluginRequestTest (no plugin
  configured): the push RPC returns UNSUPPORTED_VERSION, heartbeats
  never solicit a push, and describe reports NOT_STORED
• GroupCoordinatorBaseRequestTest: new
  streamsGroupTopologyDescriptionUpdate helper and
  IncludeTopologyDescription support on streamsGroupDescribe

End-to-end tests with real KafkaStreams clients

(streams/integration-tests)

• TopologyDescriptionPluginIntegrationTest — 4 tests covering the
  core push lifecycle on a live cluster:
  • topology description reaches the plugin after join
  • two-member group deduplicates: only one push happens
  • permanent plugin failure ratchets the failed epoch and stops
    re-solicitation (call count stays at one, the application keeps running,
    describe reports NOT_STORED)
  • explicit DeleteGroups triggers plugin.deleteTopology
• TopologyDescriptionPluginExpirationIntegrationTest — 2 tests
  covering natural group expiry:
  • group with stored topology: the cleanup cycle calls
    plugin.deleteTopology and clears the stored epoch before the
    expiration sweep tombstones the group
  • group whose push permanently failed (stored epoch -1): the
    tombstone fires directly without any deleteTopology call
• TopologyDescriptionNoPluginIntegrationTest — 1 test keeping the
  plugin-less production default exercised end-to-end at the streams
  level: a real KafkaStreams client joins and runs normally against a
  broker without a plugin (no push is ever solicited), and describe with
  includeTopologyDescription reports NOT_STORED
• The plugin-configured tests observe the broker through a new
  TrackingTopologyDescriptionPlugin test plugin (extends
  InMemoryTopologyDescriptionPlugin, records per-group
  setTopology/deleteTopology calls in static state and can inject
  permanent failures) — the embedded brokers run in the test JVM

CLI and Admin API

• DescribeStreamsGroupTest#testDescribeStreamsGroupWithTopologyOption:
  exercises kafka-streams-groups.sh --describe --topology end-to-end,
  waiting until the output contains Topologies: and the source topic
  name

PlaintextAdminIntegrationTest#testDescribeStreamsGroupsWithTopologyDescription:
DescribeStreamsGroupsOptions.includeTopologyDescription wiring
(NOT_REQUESTED / NOT_STORED on a plugin-less cluster; the
AVAILABLE path is covered via the Admin API in
TopologyDescriptionPluginIntegrationTest)

Infrastructure

• EmbeddedKafkaCluster.addDefaultBrokerPropsIfAbsent now configures
  InMemoryTopologyDescriptionPlugin via putIfAbsent, so all streams
  integration tests exercise the topology description path and catch
  regressions without per-test changes. An explicitly empty value for
  group.streams.topology.description.plugin.class opts out and runs the
  broker with the plugin-less production default (a Properties cannot
  carry null and an empty class name would fail config parsing)
• streams:integration-tests gains a group-coordinator-api test
  dependency; import-control allows org.apache.kafka.server.streams in
  streams integration tests

Testing

All 15 new tests pass locally, plus checkstyle on the touched modules.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

🤖 Generated with Claude Code

[frankvicky]

Thanks
LGTM