This is an automated email from the ASF dual-hosted git repository.
frankvicky pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 88425232ded KAFKA-20627: Add broker integration tests for topology
description plugin (#22775)
88425232ded is described below
commit 88425232ded74367a6b0fb94d31ac6ca233ec992
Author: Alieh Saeedi <[email protected]>
AuthorDate: Tue Jul 14 10:13:49 2026 +0200
KAFKA-20627: Add broker integration tests for topology description plugin
(#22775)
Adds the integration test coverage for the streams group topology
description plugin (KIP-1331) requested in
[KAFKA-20627](https://issues.apache.org/jira/browse/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
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Fable 5 <[email protected]>
---
build.gradle | 1 +
checkstyle/import-control.xml | 1 +
.../kafka/api/PlaintextAdminIntegrationTest.scala | 48 +++
.../server/GroupCoordinatorBaseRequestTest.scala | 26 +-
...oupTopologyDescriptionNoPluginRequestTest.scala | 142 +++++++++
...treamsGroupTopologyDescriptionRequestTest.scala | 352 +++++++++++++++++++++
...TopologyDescriptionNoPluginIntegrationTest.java | 116 +++++++
...DescriptionPluginExpirationIntegrationTest.java | 189 +++++++++++
.../TopologyDescriptionPluginIntegrationTest.java | 261 +++++++++++++++
.../integration/utils/EmbeddedKafkaCluster.java | 16 +
.../utils/TrackingTopologyDescriptionPlugin.java | 82 +++++
.../tools/streams/DescribeStreamsGroupTest.java | 26 ++
12 files changed, 1258 insertions(+), 2 deletions(-)
diff --git a/build.gradle b/build.gradle
index 1bd566ccee4..7fe163a9245 100644
--- a/build.gradle
+++ b/build.gradle
@@ -3065,6 +3065,7 @@ project(':streams:integration-tests') {
testImplementation testFixtures(project(':clients'))
testImplementation project(':group-coordinator')
+ testImplementation project(':group-coordinator:group-coordinator-api')
testImplementation project(':server')
testImplementation project(':server-common')
testImplementation testFixtures(project(':server-common'))
diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml
index 9062dc0f199..3a0b41cabd1 100644
--- a/checkstyle/import-control.xml
+++ b/checkstyle/import-control.xml
@@ -422,6 +422,7 @@
<allow
class="org.apache.kafka.coordinator.transaction.TransactionLogConfig" />
<allow pkg="org.apache.kafka.coordinator.group" />
<allow pkg="org.apache.kafka.network" />
+ <allow pkg="org.apache.kafka.server.streams" />
</subpackage>
<subpackage name="test">
diff --git
a/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
b/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
index 913aec618a2..5b92f13c74c 100644
---
a/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
+++
b/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
@@ -4484,6 +4484,54 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
}
}
+ @Test
+ def testDescribeStreamsGroupsWithTopologyDescription(): Unit = {
+ val streamsGroupId = "stream_group_id"
+ val testTopicName = "test_topic"
+ val testNumPartitions = 1
+
+ val config = createConfig
+ client = Admin.create(config)
+
+ prepareTopics(List(testTopicName), testNumPartitions)
+ prepareRecords(testTopicName)
+
+ val streams = createStreamsGroup(
+ inputTopics = Set(testTopicName),
+ changelogTopics = Set(testTopicName + "-changelog"),
+ streamsGroupId = streamsGroupId
+ )
+ streams.poll(JDuration.ofMillis(500L))
+
+ try {
+ TestUtils.waitUntilTrue(() => {
+ val firstGroup = client.listGroups().all().get().stream()
+ .filter(g => g.groupId() == streamsGroupId).findFirst().orElse(null)
+ firstGroup != null && firstGroup.groupState().orElse(null) ==
GroupState.STABLE
+ }, "Streams group did not transition to STABLE before timeout")
+
+ // Without IncludeTopologyDescription the status stays at its
NOT_REQUESTED default
+ // and no topology description is attached.
+ val groupWithoutTopology =
client.describeStreamsGroups(util.List.of(streamsGroupId)).all().get().get(streamsGroupId)
+ assertNotNull(groupWithoutTopology)
+ assertEquals(StreamsGroupTopologyDescriptionStatus.NOT_REQUESTED,
groupWithoutTopology.topologyDescriptionStatus())
+ assertTrue(groupWithoutTopology.topologyDescription().isEmpty)
+
+ // With IncludeTopologyDescription the brokers of this cluster have no
topology
+ // description plugin configured, so there is no description to serve:
NOT_STORED.
+ val groupWithTopology = client.describeStreamsGroups(
+ util.List.of(streamsGroupId),
+ new DescribeStreamsGroupsOptions().includeTopologyDescription(true)
+ ).all().get().get(streamsGroupId)
+ assertNotNull(groupWithTopology)
+ assertEquals(StreamsGroupTopologyDescriptionStatus.NOT_STORED,
groupWithTopology.topologyDescriptionStatus())
+ assertTrue(groupWithTopology.topologyDescription().isEmpty)
+ } finally {
+ Utils.closeQuietly(streams, "streams")
+ Utils.closeQuietly(client, "adminClient")
+ }
+ }
+
@Test
def testDescribeStreamsGroupsNotReady(): Unit = {
val streamsGroupId = "stream_group_id"
diff --git
a/core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala
b/core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala
index 307a3ae1a5e..2420ad149d6 100644
---
a/core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala
+++
b/core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala
@@ -26,9 +26,9 @@ import
org.apache.kafka.common.message.LeaveGroupRequestData.MemberIdentity
import org.apache.kafka.common.message.LeaveGroupResponseData.MemberResponse
import
org.apache.kafka.common.message.SyncGroupRequestData.SyncGroupRequestAssignment
import
org.apache.kafka.common.message.WriteTxnMarkersRequestData.{WritableTxnMarker,
WritableTxnMarkerTopic}
-import org.apache.kafka.common.message.{AddOffsetsToTxnRequestData,
AddOffsetsToTxnResponseData, ConsumerGroupDescribeRequestData,
ConsumerGroupDescribeResponseData, ConsumerGroupHeartbeatRequestData,
ConsumerGroupHeartbeatResponseData, DeleteGroupsRequestData,
DeleteGroupsResponseData, DescribeGroupsRequestData,
DescribeGroupsResponseData, EndTxnRequestData, HeartbeatRequestData,
HeartbeatResponseData, InitProducerIdRequestData, JoinGroupRequestData,
JoinGroupResponseData, LeaveGroupRes [...]
+import org.apache.kafka.common.message.{AddOffsetsToTxnRequestData,
AddOffsetsToTxnResponseData, ConsumerGroupDescribeRequestData,
ConsumerGroupDescribeResponseData, ConsumerGroupHeartbeatRequestData,
ConsumerGroupHeartbeatResponseData, DeleteGroupsRequestData,
DeleteGroupsResponseData, DescribeGroupsRequestData,
DescribeGroupsResponseData, EndTxnRequestData, HeartbeatRequestData,
HeartbeatResponseData, InitProducerIdRequestData, JoinGroupRequestData,
JoinGroupResponseData, LeaveGroupRes [...]
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
-import org.apache.kafka.common.requests.{AbstractRequest, AbstractResponse,
AddOffsetsToTxnRequest, AddOffsetsToTxnResponse, ConsumerGroupDescribeRequest,
ConsumerGroupDescribeResponse, ConsumerGroupHeartbeatRequest,
ConsumerGroupHeartbeatResponse, DeleteGroupsRequest, DeleteGroupsResponse,
DescribeGroupsRequest, DescribeGroupsResponse, EndTxnRequest, EndTxnResponse,
HeartbeatRequest, HeartbeatResponse, InitProducerIdRequest,
InitProducerIdResponse, JoinGroupRequest, JoinGroupResponse, L [...]
+import org.apache.kafka.common.requests.{AbstractRequest, AbstractResponse,
AddOffsetsToTxnRequest, AddOffsetsToTxnResponse, ConsumerGroupDescribeRequest,
ConsumerGroupDescribeResponse, ConsumerGroupHeartbeatRequest,
ConsumerGroupHeartbeatResponse, DeleteGroupsRequest, DeleteGroupsResponse,
DescribeGroupsRequest, DescribeGroupsResponse, EndTxnRequest, EndTxnResponse,
HeartbeatRequest, HeartbeatResponse, InitProducerIdRequest,
InitProducerIdResponse, JoinGroupRequest, JoinGroupResponse, L [...]
import org.apache.kafka.common.serialization.StringSerializer
import org.apache.kafka.common.test.ClusterInstance
import org.apache.kafka.common.utils.internals.ProducerIdAndEpoch
@@ -777,18 +777,40 @@ class GroupCoordinatorBaseRequestTest(cluster:
ClusterInstance) {
protected def streamsGroupDescribe(
groupIds: List[String],
includeAuthorizedOperations: Boolean = false,
+ includeTopologyDescription: Boolean = false,
version: Short =
ApiKeys.STREAMS_GROUP_DESCRIBE.latestVersion(isUnstableApiEnabled)
): List[StreamsGroupDescribeResponseData.DescribedGroup] = {
val streamsGroupDescribeRequest = new StreamsGroupDescribeRequest.Builder(
new StreamsGroupDescribeRequestData()
.setGroupIds(groupIds.asJava)
.setIncludeAuthorizedOperations(includeAuthorizedOperations)
+ .setIncludeTopologyDescription(includeTopologyDescription)
).build(version)
val streamsGroupDescribeResponse =
connectAndReceive[StreamsGroupDescribeResponse](streamsGroupDescribeRequest)
streamsGroupDescribeResponse.data.groups.asScala.toList
}
+ protected def streamsGroupTopologyDescriptionUpdate(
+ groupId: String,
+ memberId: String,
+ topologyEpoch: Int,
+ topologyDescription:
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription,
+ version: Short =
ApiKeys.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_UPDATE.latestVersion(isUnstableApiEnabled)
+ ): StreamsGroupTopologyDescriptionUpdateResponseData = {
+ val streamsGroupTopologyDescriptionUpdateRequest = new
StreamsGroupTopologyDescriptionUpdateRequest.Builder(
+ new StreamsGroupTopologyDescriptionUpdateRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId)
+ .setTopologyEpoch(topologyEpoch)
+ .setTopologyDescription(topologyDescription)
+ ).build(version)
+
+ val streamsGroupTopologyDescriptionUpdateResponse =
+
connectAndReceive[StreamsGroupTopologyDescriptionUpdateResponse](streamsGroupTopologyDescriptionUpdateRequest)
+ streamsGroupTopologyDescriptionUpdateResponse.data
+ }
+
protected def heartbeat(
groupId: String,
generationId: Int,
diff --git
a/core/src/test/scala/unit/kafka/server/StreamsGroupTopologyDescriptionNoPluginRequestTest.scala
b/core/src/test/scala/unit/kafka/server/StreamsGroupTopologyDescriptionNoPluginRequestTest.scala
new file mode 100644
index 00000000000..7e85f30f2f8
--- /dev/null
+++
b/core/src/test/scala/unit/kafka/server/StreamsGroupTopologyDescriptionNoPluginRequestTest.scala
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.server
+
+import kafka.utils.TestUtils
+import org.apache.kafka.common.message.{StreamsGroupHeartbeatRequestData,
StreamsGroupTopologyDescriptionUpdateRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.StreamsGroupDescribeResponse
+import org.apache.kafka.common.test.ClusterInstance
+import org.apache.kafka.common.test.api.{ClusterConfigProperty, ClusterTest,
ClusterTestDefaults, Type}
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertNull}
+
+import scala.jdk.CollectionConverters._
+
+/**
+ * Broker-level tests of the StreamsGroupTopologyDescriptionUpdate RPC when no
topology
+ * description plugin is configured
(`group.streams.topology.description.plugin.class`
+ * unset): the RPC is rejected with UNSUPPORTED_VERSION, heartbeats never
solicit a push,
+ * and describe reports NOT_STORED. See
[[StreamsGroupTopologyDescriptionRequestTest]]
+ * for the behavior with a plugin configured.
+ */
+@ClusterTestDefaults(
+ types = Array(Type.KRAFT),
+ serverProperties = Array(
+ new ClusterConfigProperty(key =
GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ new ClusterConfigProperty(key =
GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),
+ new ClusterConfigProperty(key =
GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, value =
"0")
+ )
+)
+class StreamsGroupTopologyDescriptionNoPluginRequestTest(cluster:
ClusterInstance) extends GroupCoordinatorBaseRequestTest(cluster) {
+
+ private val topologyEpoch = 1
+
+ @ClusterTest
+ def
testStreamsGroupTopologyDescriptionUpdateReturnsUnsupportedVersionWhenNoPluginConfigured():
Unit = {
+ // The plugin gate is checked before any group lookup, so no group setup
is needed.
+ val response = streamsGroupTopologyDescriptionUpdate(
+ groupId = "test-group",
+ memberId = "test-member",
+ topologyEpoch = topologyEpoch,
+ topologyDescription = createTopologyDescription("test-topic")
+ )
+ assertEquals(Errors.UNSUPPORTED_VERSION.code(), response.errorCode())
+ assertEquals("The broker has no streams group topology description plugin
configured.", response.errorMessage())
+ }
+
+ @ClusterTest
+ def
testHeartbeatDoesNotSolicitPushAndDescribeReturnsNotStoredWhenNoPluginConfigured():
Unit = {
+ val admin = cluster.admin()
+ val groupId = "test-group"
+ val memberId = "test-member"
+ val topicName = "test-topic"
+
+ try {
+ TestUtils.createOffsetsTopicWithAdmin(
+ admin = admin,
+ brokers = cluster.brokers.values().asScala.toSeq,
+ controllers = cluster.controllers().values().asScala.toSeq
+ )
+ TestUtils.createTopicWithAdmin(
+ admin = admin,
+ brokers = cluster.brokers.values().asScala.toSeq,
+ controllers = cluster.controllers().values().asScala.toSeq,
+ topic = topicName,
+ numPartitions = 3
+ )
+
+ // Join the group and wait until the topology epoch is resolved (tasks
assigned).
+ // No heartbeat along the way may solicit a topology description push.
+ TestUtils.waitUntilTrue(() => {
+ val response = streamsGroupHeartbeat(
+ groupId = groupId,
+ memberId = memberId,
+ rebalanceTimeoutMs = 1000,
+ activeTasks = List.empty,
+ standbyTasks = List.empty,
+ warmupTasks = List.empty,
+ topology = createMockTopology(topicName)
+ )
+ assertFalse(response.topologyDescriptionRequired(),
+ "Broker must not solicit a topology description push when no plugin
is configured.")
+ response.errorCode == Errors.NONE.code() && response.activeTasks() !=
null
+ }, "StreamsGroupHeartbeatRequest did not succeed within the timeout
period.")
+
+ // Describe with IncludeTopologyDescription: without a plugin the broker
has no
+ // description to serve, so the status is NOT_STORED.
+ val describedGroup = streamsGroupDescribe(
+ groupIds = List(groupId),
+ includeTopologyDescription = true
+ ).head
+ assertEquals(Errors.NONE.code(), describedGroup.errorCode())
+
assertEquals(StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_NOT_STORED,
describedGroup.topologyDescriptionStatus())
+ assertNull(describedGroup.topologyDescription())
+ } finally {
+ admin.close()
+ }
+ }
+
+ private def createMockTopology(topicName: String):
StreamsGroupHeartbeatRequestData.Topology = {
+ new StreamsGroupHeartbeatRequestData.Topology()
+ .setEpoch(topologyEpoch)
+ .setSubtopologies(List(
+ new StreamsGroupHeartbeatRequestData.Subtopology()
+ .setSubtopologyId("subtopology-1")
+ .setSourceTopics(List(topicName).asJava)
+ .setRepartitionSinkTopics(List.empty.asJava)
+ .setRepartitionSourceTopics(List.empty.asJava)
+ ).asJava)
+ }
+
+ private def createTopologyDescription(topicName: String):
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription = {
+ new StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription()
+ .setSubtopologies(List(
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionSubtopology()
+ .setSubtopologyId("subtopology-1")
+ .setNodes(List(
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode()
+ .setName("KSTREAM-SOURCE-0000000000")
+ .setNodeType(1.toByte) // SOURCE
+ .setSourceTopics(List(topicName).asJava)
+ .setStores(List.empty.asJava)
+ .setSuccessors(List.empty.asJava)
+ ).asJava)
+ ).asJava)
+ .setGlobalStores(List.empty.asJava)
+ }
+}
diff --git
a/core/src/test/scala/unit/kafka/server/StreamsGroupTopologyDescriptionRequestTest.scala
b/core/src/test/scala/unit/kafka/server/StreamsGroupTopologyDescriptionRequestTest.scala
new file mode 100644
index 00000000000..8352339f2ce
--- /dev/null
+++
b/core/src/test/scala/unit/kafka/server/StreamsGroupTopologyDescriptionRequestTest.scala
@@ -0,0 +1,352 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.server
+
+import kafka.utils.TestUtils
+import org.apache.kafka.common.errors.UnsupportedVersionException
+import org.apache.kafka.common.message.{StreamsGroupHeartbeatRequestData,
StreamsGroupTopologyDescriptionUpdateRequestData}
+import org.apache.kafka.common.protocol.{ApiKeys, Errors}
+import org.apache.kafka.common.requests.StreamsGroupDescribeResponse
+import org.apache.kafka.common.test.ClusterInstance
+import org.apache.kafka.common.test.api.{ClusterConfigProperty, ClusterTest,
ClusterTestDefaults, Type}
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse,
assertNotNull, assertNull, assertThrows}
+
+import scala.jdk.CollectionConverters._
+
+/**
+ * Broker-level tests of the StreamsGroupTopologyDescriptionUpdate RPC with a
topology
+ * description plugin configured
([[org.apache.kafka.server.streams.InMemoryTopologyDescriptionPlugin]]).
+ * See [[StreamsGroupTopologyDescriptionNoPluginRequestTest]] for the
plugin-less
+ * UNSUPPORTED_VERSION behavior.
+ */
+@ClusterTestDefaults(
+ types = Array(Type.KRAFT),
+ serverProperties = Array(
+ new ClusterConfigProperty(key =
GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ new ClusterConfigProperty(key =
GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),
+ new ClusterConfigProperty(key =
GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, value =
"0"),
+ new ClusterConfigProperty(
+ key =
GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG,
+ value =
"org.apache.kafka.server.streams.InMemoryTopologyDescriptionPlugin")
+ )
+)
+class StreamsGroupTopologyDescriptionRequestTest(cluster: ClusterInstance)
extends GroupCoordinatorBaseRequestTest(cluster) {
+
+ private val topologyEpoch = 1
+
+ @ClusterTest
+ def testStreamsGroupTopologyDescriptionUpdateWithInvalidApiVersion(): Unit =
{
+ assertThrows(classOf[UnsupportedVersionException], () =>
+ streamsGroupTopologyDescriptionUpdate(
+ groupId = "test-group",
+ memberId = "test-member",
+ topologyEpoch = topologyEpoch,
+ topologyDescription = createTopologyDescription("test-topic"),
+ version = -1)
+ )
+ }
+
+ @ClusterTest
+ def testHeartbeatSolicitsPushAndDescribeReturnsStoredTopologyDescription():
Unit = {
+ val admin = cluster.admin()
+ val groupId = "test-group"
+ val memberId = "test-member"
+ val topicName = "test-topic"
+
+ try {
+ TestUtils.createOffsetsTopicWithAdmin(
+ admin = admin,
+ brokers = cluster.brokers.values().asScala.toSeq,
+ controllers = cluster.controllers().values().asScala.toSeq
+ )
+ TestUtils.createTopicWithAdmin(
+ admin = admin,
+ brokers = cluster.brokers.values().asScala.toSeq,
+ controllers = cluster.controllers().values().asScala.toSeq,
+ topic = topicName,
+ numPartitions = 3
+ )
+
+ // Join the group and wait until the broker solicits a topology
description push.
+ // The flag is only set on the first heartbeat after the group's
topology epoch is
+ // resolved (the per-group back-off is armed at the same time), so it
must be
+ // accumulated across heartbeats rather than asserted on the last
response.
+ var solicited = false
+ var memberEpoch = 0
+ TestUtils.waitUntilTrue(() => {
+ val response = streamsGroupHeartbeat(
+ groupId = groupId,
+ memberId = memberId,
+ rebalanceTimeoutMs = 1000,
+ activeTasks = List.empty,
+ standbyTasks = List.empty,
+ warmupTasks = List.empty,
+ topology = createMockTopology(topicName)
+ )
+ solicited ||= response.topologyDescriptionRequired()
+ memberEpoch = response.memberEpoch()
+ response.errorCode == Errors.NONE.code() && solicited
+ }, "Broker did not solicit a topology description push within the
timeout period.")
+
+ // Push the topology description for the current topology epoch.
+ val updateResponse = streamsGroupTopologyDescriptionUpdate(
+ groupId = groupId,
+ memberId = memberId,
+ topologyEpoch = topologyEpoch,
+ topologyDescription = createTopologyDescription(topicName)
+ )
+ assertEquals(Errors.NONE.code(), updateResponse.errorCode(),
s"Unexpected error: ${updateResponse.errorMessage()}")
+
+ // Describe with IncludeTopologyDescription: the stored description must
be returned.
+ val describedGroup = streamsGroupDescribe(
+ groupIds = List(groupId),
+ includeTopologyDescription = true
+ ).head
+ assertEquals(Errors.NONE.code(), describedGroup.errorCode())
+
assertEquals(StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_AVAILABLE,
describedGroup.topologyDescriptionStatus())
+ assertNotNull(describedGroup.topologyDescription())
+ assertEquals(1,
describedGroup.topologyDescription().subtopologies().size())
+ val subtopology =
describedGroup.topologyDescription().subtopologies().get(0)
+ assertEquals("subtopology-1", subtopology.subtopologyId())
+ assertEquals(1, subtopology.nodes().size())
+ assertEquals("KSTREAM-SOURCE-0000000000",
subtopology.nodes().get(0).name())
+ assertEquals(List(topicName).asJava,
subtopology.nodes().get(0).sourceTopics())
+
+ // Describe without IncludeTopologyDescription: the status stays at its
+ // NOT_REQUESTED default and no description is attached.
+ val describedGroupWithoutTopology = streamsGroupDescribe(
+ groupIds = List(groupId)
+ ).head
+ assertEquals(Errors.NONE.code(),
describedGroupWithoutTopology.errorCode())
+
assertEquals(StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_NOT_REQUESTED,
describedGroupWithoutTopology.topologyDescriptionStatus())
+ assertNull(describedGroupWithoutTopology.topologyDescription())
+
+ // Once the description is stored at the current topology epoch,
heartbeats must not
+ // solicit another push.
+ val heartbeatAfterPush = streamsGroupHeartbeat(
+ groupId = groupId,
+ memberId = memberId,
+ memberEpoch = memberEpoch,
+ rebalanceTimeoutMs = 1000,
+ activeTasks = List.empty,
+ standbyTasks = List.empty,
+ warmupTasks = List.empty
+ )
+ assertFalse(heartbeatAfterPush.topologyDescriptionRequired(),
+ "Broker must not re-solicit a topology description push once it is
stored at the current epoch.")
+ } finally {
+ admin.close()
+ }
+ }
+
+ @ClusterTest
+ def testStreamsGroupTopologyDescriptionUpdateValidationErrors(): Unit = {
+ val admin = cluster.admin()
+ val groupId = "test-group"
+ val memberId = "test-member"
+ val topicName = "test-topic"
+
+ try {
+ TestUtils.createOffsetsTopicWithAdmin(
+ admin = admin,
+ brokers = cluster.brokers.values().asScala.toSeq,
+ controllers = cluster.controllers().values().asScala.toSeq
+ )
+ TestUtils.createTopicWithAdmin(
+ admin = admin,
+ brokers = cluster.brokers.values().asScala.toSeq,
+ controllers = cluster.controllers().values().asScala.toSeq,
+ topic = topicName,
+ numPartitions = 3
+ )
+
+ // Join the group so the coordinator is loaded and the group exists.
+ TestUtils.waitUntilTrue(() => {
+ val response = streamsGroupHeartbeat(
+ groupId = groupId,
+ memberId = memberId,
+ rebalanceTimeoutMs = 1000,
+ activeTasks = List.empty,
+ standbyTasks = List.empty,
+ warmupTasks = List.empty,
+ topology = createMockTopology(topicName)
+ )
+ response.errorCode == Errors.NONE.code()
+ }, "StreamsGroupHeartbeatRequest did not succeed within the timeout
period.")
+
+ // Empty group id.
+ var response = streamsGroupTopologyDescriptionUpdate(
+ groupId = "",
+ memberId = memberId,
+ topologyEpoch = topologyEpoch,
+ topologyDescription = createTopologyDescription(topicName)
+ )
+ assertEquals(Errors.INVALID_REQUEST.code(), response.errorCode())
+ assertEquals("GroupId can't be empty.", response.errorMessage())
+
+ // Empty member id.
+ response = streamsGroupTopologyDescriptionUpdate(
+ groupId = groupId,
+ memberId = "",
+ topologyEpoch = topologyEpoch,
+ topologyDescription = createTopologyDescription(topicName)
+ )
+ assertEquals(Errors.INVALID_REQUEST.code(), response.errorCode())
+ assertEquals("MemberId can't be empty.", response.errorMessage())
+
+ // Unknown group.
+ response = streamsGroupTopologyDescriptionUpdate(
+ groupId = "unknown-group",
+ memberId = memberId,
+ topologyEpoch = topologyEpoch,
+ topologyDescription = createTopologyDescription(topicName)
+ )
+ assertEquals(Errors.GROUP_ID_NOT_FOUND.code(), response.errorCode())
+
+ // Unknown member.
+ response = streamsGroupTopologyDescriptionUpdate(
+ groupId = groupId,
+ memberId = "unknown-member",
+ topologyEpoch = topologyEpoch,
+ topologyDescription = createTopologyDescription(topicName)
+ )
+ assertEquals(Errors.UNKNOWN_MEMBER_ID.code(), response.errorCode())
+
+ // Stale topology epoch.
+ response = streamsGroupTopologyDescriptionUpdate(
+ groupId = groupId,
+ memberId = memberId,
+ topologyEpoch = 42,
+ topologyDescription = createTopologyDescription(topicName)
+ )
+ assertEquals(Errors.INVALID_REQUEST.code(), response.errorCode())
+ assertEquals(s"Topology epoch 42 does not match the group's current
topology epoch $topologyEpoch.", response.errorMessage())
+ } finally {
+ admin.close()
+ }
+ }
+
+ @ClusterTest
+ def testDeleteGroupWithStoredTopologyDescription(): Unit = {
+ val admin = cluster.admin()
+ val groupId = "test-group"
+ val memberId = "test-member"
+ val topicName = "test-topic"
+
+ try {
+ TestUtils.createOffsetsTopicWithAdmin(
+ admin = admin,
+ brokers = cluster.brokers.values().asScala.toSeq,
+ controllers = cluster.controllers().values().asScala.toSeq
+ )
+ TestUtils.createTopicWithAdmin(
+ admin = admin,
+ brokers = cluster.brokers.values().asScala.toSeq,
+ controllers = cluster.controllers().values().asScala.toSeq,
+ topic = topicName,
+ numPartitions = 3
+ )
+
+ // Join and push the topology description.
+ TestUtils.waitUntilTrue(() => {
+ val response = streamsGroupHeartbeat(
+ groupId = groupId,
+ memberId = memberId,
+ rebalanceTimeoutMs = 1000,
+ activeTasks = List.empty,
+ standbyTasks = List.empty,
+ warmupTasks = List.empty,
+ topology = createMockTopology(topicName)
+ )
+ response.errorCode == Errors.NONE.code()
+ }, "StreamsGroupHeartbeatRequest did not succeed within the timeout
period.")
+
+ val updateResponse = streamsGroupTopologyDescriptionUpdate(
+ groupId = groupId,
+ memberId = memberId,
+ topologyEpoch = topologyEpoch,
+ topologyDescription = createTopologyDescription(topicName)
+ )
+ assertEquals(Errors.NONE.code(), updateResponse.errorCode(),
s"Unexpected error: ${updateResponse.errorMessage()}")
+
+ val describedGroup = streamsGroupDescribe(
+ groupIds = List(groupId),
+ includeTopologyDescription = true
+ ).head
+
assertEquals(StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_AVAILABLE,
describedGroup.topologyDescriptionStatus())
+
+ // The member leaves so the group becomes empty and can be deleted.
+ val leaveResponse = streamsGroupHeartbeat(
+ groupId = groupId,
+ memberId = memberId,
+ memberEpoch = -1,
+ rebalanceTimeoutMs = 1000,
+ activeTasks = List.empty,
+ standbyTasks = List.empty,
+ warmupTasks = List.empty
+ )
+ assertEquals(Errors.NONE.code(), leaveResponse.errorCode())
+
+ // Deleting the group drives plugin.deleteTopology before the tombstone;
a NONE
+ // result (rather than GROUP_DELETION_FAILED) shows the plugin delete
succeeded.
+ deleteGroups(
+ groupIds = List(groupId),
+ expectedErrors = List(Errors.NONE),
+ version = ApiKeys.DELETE_GROUPS.latestVersion(isUnstableApiEnabled)
+ )
+
+ // The group is gone.
+ val describedAfterDelete = streamsGroupDescribe(
+ groupIds = List(groupId),
+ includeTopologyDescription = true
+ ).head
+ assertEquals(Errors.GROUP_ID_NOT_FOUND.code(),
describedAfterDelete.errorCode())
+ } finally {
+ admin.close()
+ }
+ }
+
+ private def createMockTopology(topicName: String):
StreamsGroupHeartbeatRequestData.Topology = {
+ new StreamsGroupHeartbeatRequestData.Topology()
+ .setEpoch(topologyEpoch)
+ .setSubtopologies(List(
+ new StreamsGroupHeartbeatRequestData.Subtopology()
+ .setSubtopologyId("subtopology-1")
+ .setSourceTopics(List(topicName).asJava)
+ .setRepartitionSinkTopics(List.empty.asJava)
+ .setRepartitionSourceTopics(List.empty.asJava)
+ ).asJava)
+ }
+
+ private def createTopologyDescription(topicName: String):
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription = {
+ new StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription()
+ .setSubtopologies(List(
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionSubtopology()
+ .setSubtopologyId("subtopology-1")
+ .setNodes(List(
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode()
+ .setName("KSTREAM-SOURCE-0000000000")
+ .setNodeType(1.toByte) // SOURCE
+ .setSourceTopics(List(topicName).asJava)
+ .setStores(List.empty.asJava)
+ .setSuccessors(List.empty.asJava)
+ ).asJava)
+ ).asJava)
+ .setGlobalStores(List.empty.asJava)
+ }
+}
diff --git
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TopologyDescriptionNoPluginIntegrationTest.java
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TopologyDescriptionNoPluginIntegrationTest.java
new file mode 100644
index 00000000000..90e4d8500fd
--- /dev/null
+++
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TopologyDescriptionNoPluginIntegrationTest.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.integration;
+
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.DescribeStreamsGroupsOptions;
+import org.apache.kafka.clients.admin.StreamsGroupDescription;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescriptionStatus;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.streams.GroupProtocol;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.streams.kstream.Consumed;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Properties;
+
+import static
org.apache.kafka.streams.integration.utils.IntegrationTestUtils.startApplicationAndWaitUntilRunning;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * End-to-end test of a real {@link KafkaStreams} client against a broker
running the
+ * plugin-less production default (no topology description plugin configured,
KIP-1331).
+ * {@link EmbeddedKafkaCluster} configures an in-memory plugin by default for
all streams
+ * integration tests, so this test opts out explicitly to keep the production
default
+ * exercised at the streams level.
+ */
+@Timeout(600)
+@Tag("integration")
+public class TopologyDescriptionNoPluginIntegrationTest {
+
+ private static EmbeddedKafkaCluster cluster;
+ private static String bootstrapServers;
+
+ @BeforeAll
+ public static void startCluster() throws IOException {
+ final Properties props = new Properties();
+
props.put(GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG,
+ EmbeddedKafkaCluster.NO_TOPOLOGY_DESCRIPTION_PLUGIN);
+ cluster = new EmbeddedKafkaCluster(1, props);
+ cluster.start();
+ bootstrapServers = cluster.bootstrapServers();
+ }
+
+ @AfterAll
+ public static void closeCluster() {
+ cluster.stop();
+ cluster = null;
+ }
+
+ @Test
+ public void
shouldRunStreamsApplicationAgainstBrokerWithoutTopologyDescriptionPlugin()
throws Exception {
+ final String appId = "topology-description-no-plugin-app";
+ final String inputTopic = "topology-description-no-plugin-input";
+ cluster.createTopic(inputTopic, 1, 1);
+
+ final Properties config = new Properties();
+ config.put(StreamsConfig.APPLICATION_ID_CONFIG, appId);
+ config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+ config.put(StreamsConfig.STATE_DIR_CONFIG,
TestUtils.tempDirectory().getPath());
+ config.put(StreamsConfig.GROUP_PROTOCOL_CONFIG,
GroupProtocol.STREAMS.name().toLowerCase(Locale.getDefault()));
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.stream(inputTopic, Consumed.with(Serdes.String(),
Serdes.String()))
+ .foreach((key, value) -> { });
+ final Topology topology = builder.build();
+
+ try (final Admin admin =
Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
bootstrapServers));
+ final KafkaStreams streams = new KafkaStreams(topology, config)) {
+ // Without a plugin the broker never solicits a topology
description push; the
+ // application must join and run normally regardless.
+ startApplicationAndWaitUntilRunning(streams);
+
+ final StreamsGroupDescription description =
admin.describeStreamsGroups(
+ List.of(appId),
+ new
DescribeStreamsGroupsOptions().includeTopologyDescription(true))
+ .describedGroups()
+ .get(appId)
+ .get();
+ assertFalse(description.members().isEmpty());
+ assertEquals(StreamsGroupTopologyDescriptionStatus.NOT_STORED,
description.topologyDescriptionStatus());
+ assertTrue(description.topologyDescription().isEmpty());
+ }
+ }
+}
diff --git
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TopologyDescriptionPluginExpirationIntegrationTest.java
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TopologyDescriptionPluginExpirationIntegrationTest.java
new file mode 100644
index 00000000000..06762f1c3fa
--- /dev/null
+++
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TopologyDescriptionPluginExpirationIntegrationTest.java
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.integration;
+
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.DescribeStreamsGroupsOptions;
+import org.apache.kafka.clients.admin.StreamsGroupDescription;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescriptionStatus;
+import org.apache.kafka.common.errors.GroupIdNotFoundException;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.streams.GroupProtocol;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import
org.apache.kafka.streams.integration.utils.TrackingTopologyDescriptionPlugin;
+import org.apache.kafka.streams.kstream.Consumed;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+
+import static
org.apache.kafka.streams.integration.utils.IntegrationTestUtils.startApplicationAndWaitUntilRunning;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Tests of the topology description plugin behavior when a streams group
expires naturally
+ * (KIP-1331): the periodic broker-side cleanup cycle calls {@code
plugin.deleteTopology} and
+ * clears the stored topology epoch for empty groups without offsets, after
which the regular
+ * expiration sweep tombstones the group. Groups whose push permanently failed
have no stored
+ * plugin state, so they are tombstoned directly without a {@code
deleteTopology} call.
+ *
+ * <p>The tests avoid producing any input data, so the groups never commit
offsets and become
+ * eligible for expiration as soon as their last member leaves.
+ */
+@Timeout(600)
+@Tag("integration")
+public class TopologyDescriptionPluginExpirationIntegrationTest {
+
+ private static EmbeddedKafkaCluster cluster;
+ private static String bootstrapServers;
+
+ @BeforeAll
+ public static void startCluster() throws IOException {
+ final Properties props = new Properties();
+ props.put(
+
GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG,
+ TrackingTopologyDescriptionPlugin.class.getName());
+ // Run the offset-expiration sweep and the topology-description
cleanup cycle (both
+ // are scheduled at this interval) frequently so expiry happens within
seconds.
+
props.put(GroupCoordinatorConfig.OFFSETS_RETENTION_CHECK_INTERVAL_MS_CONFIG,
"1000");
+
props.put(GroupCoordinatorConfig.STREAMS_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG,
"200");
+ cluster = new EmbeddedKafkaCluster(1, props);
+ cluster.start();
+ bootstrapServers = cluster.bootstrapServers();
+ }
+
+ @AfterAll
+ public static void closeCluster() {
+ cluster.stop();
+ cluster = null;
+ }
+
+ @BeforeEach
+ public void resetPlugin() {
+ TrackingTopologyDescriptionPlugin.reset();
+ }
+
+ @Test
+ public void shouldDeleteStoredTopologyDescriptionWhenGroupExpires() throws
Exception {
+ final String appId = "topology-description-expiration-app";
+ final String inputTopic = "topology-description-expiration-input";
+ cluster.createTopic(inputTopic, 1, 1);
+
+ try (final Admin admin = createAdmin()) {
+ try (final KafkaStreams streams = new
KafkaStreams(topology(inputTopic), streamsConfig(appId))) {
+ startApplicationAndWaitUntilRunning(streams);
+ TestUtils.waitForCondition(
+ () -> describeGroup(admin,
appId).topologyDescriptionStatus() ==
StreamsGroupTopologyDescriptionStatus.AVAILABLE,
+ "Expected the topology description of group " + appId + "
to become AVAILABLE");
+ }
+
+ // The member has left and the group has no committed offsets, so
the cleanup
+ // cycle must call plugin.deleteTopology and clear the stored
epoch, after which
+ // the expiration sweep tombstones the group.
+ TestUtils.waitForCondition(
+ () ->
TrackingTopologyDescriptionPlugin.deleteTopologyCalls(appId) >= 1,
+ "Expected the cleanup cycle to invoke deleteTopology for group
" + appId);
+ waitForGroupToBeDeleted(admin, appId);
+ }
+ }
+
+ @Test
+ public void
shouldExpireGroupWithoutDeleteTopologyWhenPushPermanentlyFailed() throws
Exception {
+ final String appId = "topology-description-failed-expiration-app";
+ final String inputTopic =
"topology-description-failed-expiration-input";
+ cluster.createTopic(inputTopic, 1, 1);
+ TrackingTopologyDescriptionPlugin.failPermanently(appId);
+
+ try (final Admin admin = createAdmin()) {
+ try (final KafkaStreams streams = new
KafkaStreams(topology(inputTopic), streamsConfig(appId))) {
+ startApplicationAndWaitUntilRunning(streams);
+ TestUtils.waitForCondition(
+ () ->
TrackingTopologyDescriptionPlugin.setTopologyCalls(appId) >= 1,
+ "Expected a setTopology call for group " + appId);
+ }
+
+ // The push permanently failed, so no topology description is
stored for the
+ // group: the tombstone fires directly, without any deleteTopology
call.
+ waitForGroupToBeDeleted(admin, appId);
+ assertEquals(0,
TrackingTopologyDescriptionPlugin.deleteTopologyCalls(appId),
+ "Expected no deleteTopology call for group " + appId + " whose
push permanently failed");
+ }
+ }
+
+ private static Topology topology(final String inputTopic) {
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.stream(inputTopic, Consumed.with(Serdes.String(),
Serdes.String()))
+ .foreach((key, value) -> { });
+ return builder.build();
+ }
+
+ private static Properties streamsConfig(final String appId) {
+ final Properties config = new Properties();
+ config.put(StreamsConfig.APPLICATION_ID_CONFIG, appId);
+ config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+ config.put(StreamsConfig.STATE_DIR_CONFIG,
TestUtils.tempDirectory().getPath());
+ config.put(StreamsConfig.GROUP_PROTOCOL_CONFIG,
GroupProtocol.STREAMS.name().toLowerCase(Locale.getDefault()));
+ return config;
+ }
+
+ private static Admin createAdmin() {
+ return Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
bootstrapServers));
+ }
+
+ private static StreamsGroupDescription describeGroup(final Admin admin,
final String groupId) throws Exception {
+ return admin.describeStreamsGroups(
+ List.of(groupId),
+ new
DescribeStreamsGroupsOptions().includeTopologyDescription(true))
+ .describedGroups()
+ .get(groupId)
+ .get();
+ }
+
+ private static void waitForGroupToBeDeleted(final Admin admin, final
String groupId) throws InterruptedException {
+ TestUtils.waitForCondition(() -> {
+ try {
+ describeGroup(admin, groupId);
+ return false;
+ } catch (final ExecutionException exception) {
+ if (exception.getCause() instanceof GroupIdNotFoundException) {
+ return true;
+ }
+ // Unexpected failures (e.g. admin timeouts) must not be
folded into
+ // "group not yet deleted": waitForCondition retries a
throwing condition
+ // and rethrows the last exception at timeout, preserving the
diagnostics.
+ throw exception;
+ }
+ }, "Expected group " + groupId + " to be tombstoned by the expiration
sweep");
+ }
+}
diff --git
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TopologyDescriptionPluginIntegrationTest.java
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TopologyDescriptionPluginIntegrationTest.java
new file mode 100644
index 00000000000..900117f1ff1
--- /dev/null
+++
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TopologyDescriptionPluginIntegrationTest.java
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.integration;
+
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.DescribeStreamsGroupsOptions;
+import org.apache.kafka.clients.admin.StreamsGroupDescription;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescription;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescriptionStatus;
+import org.apache.kafka.common.errors.GroupIdNotFoundException;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.streams.GroupProtocol;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import
org.apache.kafka.streams.integration.utils.TrackingTopologyDescriptionPlugin;
+import org.apache.kafka.streams.kstream.Consumed;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+
+import static
org.apache.kafka.streams.integration.utils.IntegrationTestUtils.startApplicationAndWaitUntilRunning;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * End-to-end tests of the topology description push lifecycle (KIP-1331) on a
live cluster:
+ * a real {@link KafkaStreams} client pushes its topology description when the
broker solicits
+ * it on a heartbeat, and the broker stores it via the configured
+ * {@link TrackingTopologyDescriptionPlugin}.
+ */
+@Timeout(600)
+@Tag("integration")
+public class TopologyDescriptionPluginIntegrationTest {
+
+ private static EmbeddedKafkaCluster cluster;
+ private static String bootstrapServers;
+
+ @BeforeAll
+ public static void startCluster() throws IOException {
+ final Properties props = new Properties();
+ props.put(
+
GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG,
+ TrackingTopologyDescriptionPlugin.class.getName());
+ // Frequent heartbeats keep the solicitation round-trips short.
+
props.put(GroupCoordinatorConfig.STREAMS_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG,
"200");
+ // Effectively disable the periodic cleanup cycle and expiration sweep
(both run at
+ // this interval, 10 min by default):
shouldInvokeDeleteTopologyOnDeleteGroups
+ // relies on the explicit DeleteGroups call being the only delete
path, and a slow
+ // run crossing the default interval could otherwise expire the empty,
offset-less
+ // group first.
+
props.put(GroupCoordinatorConfig.OFFSETS_RETENTION_CHECK_INTERVAL_MS_CONFIG,
String.valueOf(Integer.MAX_VALUE));
+ cluster = new EmbeddedKafkaCluster(1, props);
+ cluster.start();
+ bootstrapServers = cluster.bootstrapServers();
+ }
+
+ @AfterAll
+ public static void closeCluster() {
+ cluster.stop();
+ cluster = null;
+ }
+
+ @BeforeEach
+ public void resetPlugin() {
+ TrackingTopologyDescriptionPlugin.reset();
+ }
+
+ @Test
+ public void shouldPushTopologyDescriptionToPluginAfterJoin() throws
Exception {
+ final String appId = "topology-description-push-app";
+ final String inputTopic = "topology-description-push-input";
+ cluster.createTopic(inputTopic, 1, 1);
+
+ try (final Admin admin = createAdmin();
+ final KafkaStreams streams = new
KafkaStreams(topology(inputTopic), streamsConfig(appId))) {
+ startApplicationAndWaitUntilRunning(streams);
+
+ waitForTopologyDescriptionAvailable(admin, appId);
+
assertTrue(TrackingTopologyDescriptionPlugin.setTopologyCalls(appId) >= 1,
+ "Expected at least one setTopology call for group " + appId);
+
+ // The stored description carries the source topic.
+ final StreamsGroupTopologyDescription topologyDescription =
+ describeGroup(admin, appId,
true).topologyDescription().orElseThrow();
+ assertEquals(1, topologyDescription.subtopologies().size());
+ final boolean sourceTopicFound =
topologyDescription.subtopologies().stream()
+ .flatMap(subtopology -> subtopology.nodes().stream())
+
.filter(StreamsGroupTopologyDescription.Source.class::isInstance)
+ .map(StreamsGroupTopologyDescription.Source.class::cast)
+ .anyMatch(source -> source.topics().contains(inputTopic));
+ assertTrue(sourceTopicFound, "Expected a source node reading from
" + inputTopic);
+
+ // Describing without requesting the topology description must not
attach one.
+ final StreamsGroupDescription withoutTopology =
describeGroup(admin, appId, false);
+ assertEquals(StreamsGroupTopologyDescriptionStatus.NOT_REQUESTED,
withoutTopology.topologyDescriptionStatus());
+ assertTrue(withoutTopology.topologyDescription().isEmpty());
+ }
+ }
+
+ @Test
+ public void shouldPushTopologyDescriptionOnlyOnceForTwoMembers() throws
Exception {
+ final String appId = "topology-description-dedup-app";
+ final String inputTopic = "topology-description-dedup-input";
+ cluster.createTopic(inputTopic, 2, 1);
+
+ try (final Admin admin = createAdmin();
+ final KafkaStreams streams1 = new
KafkaStreams(topology(inputTopic), streamsConfig(appId));
+ final KafkaStreams streams2 = new
KafkaStreams(topology(inputTopic), streamsConfig(appId))) {
+ startApplicationAndWaitUntilRunning(List.of(streams1, streams2));
+
+ waitForTopologyDescriptionAvailable(admin, appId);
+ TestUtils.waitForCondition(
+ () -> describeGroup(admin, appId, false).members().size() == 2,
+ "Expected two members in group " + appId);
+
+ // The broker solicits the push from a single member at a time and
stops
+ // soliciting once the description is stored, so a two-member
group must
+ // trigger exactly one setTopology call. Accepted flake risk: the
client
+ // retries the push on a retriable send failure (e.g. a disconnect
after the
+ // broker already processed the request), which would produce a
duplicate
+ // setTopology call at the same epoch — client-retry noise, not a
dedup
+ // regression. If this assertion ever flakes, that is why.
+ assertEquals(1,
TrackingTopologyDescriptionPlugin.setTopologyCalls(appId),
+ "Expected exactly one setTopology call for group " + appId);
+ }
+ }
+
+ @Test
+ public void shouldStopSolicitingPushAfterPermanentPluginFailure() throws
Exception {
+ final String appId = "topology-description-failure-app";
+ final String inputTopic = "topology-description-failure-input";
+ cluster.createTopic(inputTopic, 1, 1);
+ TrackingTopologyDescriptionPlugin.failPermanently(appId);
+
+ try (final Admin admin = createAdmin();
+ final KafkaStreams streams = new
KafkaStreams(topology(inputTopic), streamsConfig(appId))) {
+ startApplicationAndWaitUntilRunning(streams);
+
+ TestUtils.waitForCondition(
+ () ->
TrackingTopologyDescriptionPlugin.setTopologyCalls(appId) >= 1,
+ "Expected a setTopology call for group " + appId);
+
+ // The permanent failure ratchets the group's failed topology
epoch, so the broker
+ // must not re-solicit the push: the call count stays at one
across many heartbeat
+ // intervals, the application keeps running, and describe reports
NOT_STORED.
+ // Note: this window is shorter than the 30s initial
re-solicitation back-off, so
+ // it cannot distinguish the permanent-failure ratchet from an
armed back-off (a
+ // permanent failure misclassified as transient would also stay at
one call here);
+ // that classification is pinned down at the unit level by
+ // GroupCoordinatorServiceTopologyDescriptionTest. What this does
catch is the
+ // ratchet being lost entirely, since re-solicitation would then
happen within a
+ // couple of heartbeats.
+ Thread.sleep(2000);
+ assertEquals(1,
TrackingTopologyDescriptionPlugin.setTopologyCalls(appId),
+ "Expected no re-solicitation after a permanent plugin failure
for group " + appId);
+ assertEquals(KafkaStreams.State.RUNNING, streams.state());
+ assertEquals(StreamsGroupTopologyDescriptionStatus.NOT_STORED,
+ describeGroup(admin, appId, true).topologyDescriptionStatus());
+ }
+ }
+
+ @Test
+ public void shouldInvokeDeleteTopologyOnDeleteGroups() throws Exception {
+ final String appId = "topology-description-delete-app";
+ final String inputTopic = "topology-description-delete-input";
+ cluster.createTopic(inputTopic, 1, 1);
+
+ try (final Admin admin = createAdmin()) {
+ try (final KafkaStreams streams = new
KafkaStreams(topology(inputTopic), streamsConfig(appId))) {
+ startApplicationAndWaitUntilRunning(streams);
+ waitForTopologyDescriptionAvailable(admin, appId);
+ }
+
+ // The application has left the group; once it is empty it can be
deleted.
+ TestUtils.waitForCondition(
+ () -> describeGroup(admin, appId, false).members().isEmpty(),
+ "Expected group " + appId + " to become empty");
+ assertEquals(0,
TrackingTopologyDescriptionPlugin.deleteTopologyCalls(appId));
+
+ admin.deleteStreamsGroups(List.of(appId)).all().get();
+
+
assertTrue(TrackingTopologyDescriptionPlugin.deleteTopologyCalls(appId) >= 1,
+ "Expected DeleteGroups to invoke deleteTopology for group " +
appId);
+ final ExecutionException exception =
assertThrows(ExecutionException.class,
+ () -> admin.describeStreamsGroups(List.of(appId)).all().get());
+ assertInstanceOf(GroupIdNotFoundException.class,
exception.getCause());
+ }
+ }
+
+ private static Topology topology(final String inputTopic) {
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.stream(inputTopic, Consumed.with(Serdes.String(),
Serdes.String()))
+ .foreach((key, value) -> { });
+ return builder.build();
+ }
+
+ private static Properties streamsConfig(final String appId) {
+ final Properties config = new Properties();
+ config.put(StreamsConfig.APPLICATION_ID_CONFIG, appId);
+ config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+ config.put(StreamsConfig.STATE_DIR_CONFIG,
TestUtils.tempDirectory().getPath());
+ config.put(StreamsConfig.GROUP_PROTOCOL_CONFIG,
GroupProtocol.STREAMS.name().toLowerCase(Locale.getDefault()));
+ return config;
+ }
+
+ private static Admin createAdmin() {
+ return Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
bootstrapServers));
+ }
+
+ private static StreamsGroupDescription describeGroup(
+ final Admin admin,
+ final String groupId,
+ final boolean includeTopologyDescription
+ ) throws Exception {
+ return admin.describeStreamsGroups(
+ List.of(groupId),
+ new
DescribeStreamsGroupsOptions().includeTopologyDescription(includeTopologyDescription))
+ .describedGroups()
+ .get(groupId)
+ .get();
+ }
+
+ private static void waitForTopologyDescriptionAvailable(final Admin admin,
final String groupId) throws InterruptedException {
+ TestUtils.waitForCondition(
+ () -> describeGroup(admin, groupId,
true).topologyDescriptionStatus() ==
StreamsGroupTopologyDescriptionStatus.AVAILABLE,
+ "Expected the topology description of group " + groupId + " to
become AVAILABLE");
+ }
+}
diff --git
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/utils/EmbeddedKafkaCluster.java
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/utils/EmbeddedKafkaCluster.java
index ba3451589f0..f2b34ec14d9 100644
---
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/utils/EmbeddedKafkaCluster.java
+++
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/utils/EmbeddedKafkaCluster.java
@@ -48,6 +48,7 @@ import
org.apache.kafka.coordinator.transaction.TransactionLogConfig;
import org.apache.kafka.network.SocketServerConfigs;
import org.apache.kafka.server.config.ServerConfigs;
import org.apache.kafka.server.config.ServerLogConfigs;
+import org.apache.kafka.server.streams.InMemoryTopologyDescriptionPlugin;
import org.apache.kafka.server.util.MockTime;
import org.apache.kafka.storage.internals.log.CleanerConfig;
import org.apache.kafka.test.TestCondition;
@@ -89,6 +90,17 @@ import static
org.apache.kafka.common.utils.Utils.mkProperties;
public class EmbeddedKafkaCluster {
private static final Logger log =
LoggerFactory.getLogger(EmbeddedKafkaCluster.class);
+
+ /**
+ * Value for {@link
GroupCoordinatorConfig#STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG}
+ * that opts a test cluster out of the {@link
InMemoryTopologyDescriptionPlugin} configured by
+ * default, running the broker with the plugin-less production default.
Needed because the
+ * config's default is null, but a {@link Properties} cannot carry null
and an empty class name
+ * would fail config parsing, so {@link #addDefaultBrokerPropsIfAbsent}
removes the property
+ * when it is set to this value.
+ */
+ public static final String NO_TOPOLOGY_DESCRIPTION_PLUGIN = "";
+
private final KafkaClusterTestKit cluster;
private final Properties brokerConfig;
public final MockTime time;
@@ -401,6 +413,10 @@ public class EmbeddedKafkaCluster {
private void addDefaultBrokerPropsIfAbsent(final Properties brokerConfig) {
brokerConfig.putIfAbsent(CleanerConfig.LOG_CLEANER_DEDUPE_BUFFER_SIZE_PROP, 2 *
1024 * 1024L);
+
brokerConfig.putIfAbsent(GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG,
InMemoryTopologyDescriptionPlugin.class.getName());
+ if
(NO_TOPOLOGY_DESCRIPTION_PLUGIN.equals(brokerConfig.get(GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG)))
{
+
brokerConfig.remove(GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG);
+ }
brokerConfig.putIfAbsent(GroupCoordinatorConfig.STREAMS_GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG,
"100");
brokerConfig.putIfAbsent(GroupCoordinatorConfig.STREAMS_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG,
"100");
brokerConfig.putIfAbsent(GroupCoordinatorConfig.GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG,
"0");
diff --git
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/utils/TrackingTopologyDescriptionPlugin.java
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/utils/TrackingTopologyDescriptionPlugin.java
new file mode 100644
index 00000000000..e6563c18ec2
--- /dev/null
+++
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/utils/TrackingTopologyDescriptionPlugin.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.integration.utils;
+
+import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription;
+import
org.apache.kafka.coordinator.group.api.streams.StreamsTopologyDescriptionPermanentFailureException;
+import org.apache.kafka.server.streams.InMemoryTopologyDescriptionPlugin;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * An {@link InMemoryTopologyDescriptionPlugin} for integration tests that
records every
+ * {@code setTopology} / {@code deleteTopology} invocation per group and can
be told to
+ * reject pushes for selected groups with a permanent failure.
+ *
+ * <p>The brokers of an embedded cluster run in the test JVM, so the recorded
state is kept
+ * in static fields the test can inspect directly. Call {@link #reset()}
between tests.
+ */
+public class TrackingTopologyDescriptionPlugin extends
InMemoryTopologyDescriptionPlugin {
+
+ private static final Map<String, AtomicInteger> SET_TOPOLOGY_CALLS = new
ConcurrentHashMap<>();
+ private static final Map<String, AtomicInteger> DELETE_TOPOLOGY_CALLS =
new ConcurrentHashMap<>();
+ private static final Set<String> FAIL_PERMANENTLY_GROUPS =
ConcurrentHashMap.newKeySet();
+
+ public static void reset() {
+ SET_TOPOLOGY_CALLS.clear();
+ DELETE_TOPOLOGY_CALLS.clear();
+ FAIL_PERMANENTLY_GROUPS.clear();
+ }
+
+ /**
+ * Make every subsequent {@code setTopology} for this group fail with a
+ * {@link StreamsTopologyDescriptionPermanentFailureException}.
+ */
+ public static void failPermanently(final String groupId) {
+ FAIL_PERMANENTLY_GROUPS.add(groupId);
+ }
+
+ public static int setTopologyCalls(final String groupId) {
+ final AtomicInteger calls = SET_TOPOLOGY_CALLS.get(groupId);
+ return calls == null ? 0 : calls.get();
+ }
+
+ public static int deleteTopologyCalls(final String groupId) {
+ final AtomicInteger calls = DELETE_TOPOLOGY_CALLS.get(groupId);
+ return calls == null ? 0 : calls.get();
+ }
+
+ @Override
+ public CompletableFuture<Void> setTopology(final String groupId, final int
topologyEpoch, final StreamsGroupTopologyDescription description) {
+ SET_TOPOLOGY_CALLS.computeIfAbsent(groupId, id -> new
AtomicInteger()).incrementAndGet();
+ if (FAIL_PERMANENTLY_GROUPS.contains(groupId)) {
+ return CompletableFuture.failedFuture(
+ new
StreamsTopologyDescriptionPermanentFailureException("Topology description
rejected by test plugin."));
+ }
+ return super.setTopology(groupId, topologyEpoch, description);
+ }
+
+ @Override
+ public CompletableFuture<Void> deleteTopology(final String groupId) {
+ DELETE_TOPOLOGY_CALLS.computeIfAbsent(groupId, id -> new
AtomicInteger()).incrementAndGet();
+ return super.deleteTopology(groupId);
+ }
+}
diff --git
a/tools/src/test/java/org/apache/kafka/tools/streams/DescribeStreamsGroupTest.java
b/tools/src/test/java/org/apache/kafka/tools/streams/DescribeStreamsGroupTest.java
index 0e62f2d7acb..9a4a16a9a20 100644
---
a/tools/src/test/java/org/apache/kafka/tools/streams/DescribeStreamsGroupTest.java
+++
b/tools/src/test/java/org/apache/kafka/tools/streams/DescribeStreamsGroupTest.java
@@ -20,6 +20,7 @@ import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.internals.Exit;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
import org.apache.kafka.streams.GroupProtocol;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
@@ -75,6 +76,12 @@ public class DescribeStreamsGroupTest {
public static void setup() throws Exception {
// start the cluster and create the input topic
final Properties props = new Properties();
+ // Configure the topology description plugin explicitly rather than
relying on
+ // EmbeddedKafkaCluster's default:
testDescribeStreamsGroupWithTopologyOption
+ // depends on it, and without a plugin the --topology option would
time out
+ // with nothing stored on the broker.
+
props.put(GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG,
+
"org.apache.kafka.server.streams.InMemoryTopologyDescriptionPlugin");
cluster = new EmbeddedKafkaCluster(1, props);
cluster.start();
cluster.createTopic(INPUT_TOPIC, 2, 1);
@@ -247,6 +254,25 @@ public class DescribeStreamsGroupTest {
}
}
+ @Test
+ public void testDescribeStreamsGroupWithTopologyOption() throws Exception {
+ final List<String> args = List.of("--bootstrap-server",
bootstrapServers, "--describe", "--topology", "--group", APP_ID);
+ final AtomicReference<String> out = new AtomicReference<>("");
+ // The streams client pushes the topology description when the broker
solicits it on
+ // a heartbeat, so retry until the broker-side plugin has it stored.
Until then the
+ // command reports that no description is stored and exits non-zero,
so the exit code
+ // is not asserted inside the retry loop. The solicit -> push -> store
round trip can
+ // take a couple of heartbeat intervals (5s each on this cluster), so
allow a timeout
+ // well above that.
+ TestUtils.waitForCondition(() -> {
+ String output = ToolsTestUtils.grabConsoleOutput(() ->
StreamsGroupCommand.execute(args.toArray(new String[0])));
+ out.set(output);
+ return output.contains("Topologies:") &&
output.contains(INPUT_TOPIC);
+ }, 60_000L, () -> String.format("Expected the topology description
with source topic %s, but found:%n%s", INPUT_TOPIC, out.get()));
+
+ assertTrue(out.get().contains("Sub-topology: 0"), "Expected a
sub-topology in the output, but found:\n" + out.get());
+ }
+
@Test
public void testDescribeNonExistingStreamsGroup() {
final String nonExistingGroup = "non-existing-group";
