You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by lu...@apache.org on 2026/10/08 10:00:53 UTC
(kafka) branch trunk updated: MINOR: Add broker integration tests for topology-description plugin failure surfacing (#23670)
This is an automated email from the ASF dual-hosted git repository.
lucasbru 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 c553733515e MINOR: Add broker integration tests for topology-description plugin failure surfacing (#23670)
c553733515e is described below
commit c553733515e35e636a1a5b3e1d45f773a1d64f02
Author: Lucas Brutschy <lb...@confluent.io>
AuthorDate: Thu Oct 8 12:00:45 2026 +0200
MINOR: Add broker integration tests for topology-description plugin failure surfacing (#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 <fr...@apache.org>, Andy Huang
<16...@users.noreply.github.com>
---
.../server/FailingTopologyDescriptionPlugin.java | 86 ++++++
...treamsGroupTopologyDescriptionRequestTest.scala | 302 ++++++++++++++++++++-
2 files changed, 385 insertions(+), 3 deletions(-)
diff --git a/core/src/test/java/kafka/server/FailingTopologyDescriptionPlugin.java b/core/src/test/java/kafka/server/FailingTopologyDescriptionPlugin.java
new file mode 100644
index 00000000000..0fae2199176
--- /dev/null
+++ b/core/src/test/java/kafka/server/FailingTopologyDescriptionPlugin.java
@@ -0,0 +1,86 @@
+/*
+ * 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 org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription;
+import org.apache.kafka.coordinator.group.api.streams.StreamsTopologyDescriptionPermanentFailureException;
+import org.apache.kafka.coordinator.group.api.streams.StreamsTopologyDescriptionTransientFailureException;
+import org.apache.kafka.server.streams.InMemoryTopologyDescriptionPlugin;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Test-only {@link org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescriptionPlugin}
+ * whose {@code setTopology} and {@code deleteTopology} failures are toggled at runtime, so broker
+ * integration tests can exercise permanent/transient plugin-failure surfacing without a real
+ * backend. Calls that are not configured to fail delegate to {@link InMemoryTopologyDescriptionPlugin},
+ * which this class extends.
+ *
+ * <p>The broker, not the test, creates the plugin instance, so the failure toggles are static.
+ * Callers must {@link #reset()} before relying on this plugin, since the JVM-wide state otherwise
+ * leaks across test methods sharing this class. Like its parent, this plugin's storage is local to
+ * one broker process, so it is only safe for the single-broker {@code ClusterTest} configurations
+ * it is used with today, not for a multi-broker cluster.
+ */
+public class FailingTopologyDescriptionPlugin extends InMemoryTopologyDescriptionPlugin {
+
+ public enum SetTopologyFailureMode {
+ NONE, PERMANENT, TRANSIENT
+ }
+
+ private static final AtomicReference<SetTopologyFailureMode> SET_TOPOLOGY_FAILURE_MODE =
+ new AtomicReference<>(SetTopologyFailureMode.NONE);
+ private static final AtomicReference<RuntimeException> DELETE_TOPOLOGY_FAILURE =
+ new AtomicReference<>(null);
+
+ public static void failNextSetTopology(SetTopologyFailureMode mode) {
+ SET_TOPOLOGY_FAILURE_MODE.set(mode);
+ }
+
+ public static void failDeleteTopologyWith(RuntimeException exception) {
+ DELETE_TOPOLOGY_FAILURE.set(exception);
+ }
+
+ public static void reset() {
+ SET_TOPOLOGY_FAILURE_MODE.set(SetTopologyFailureMode.NONE);
+ DELETE_TOPOLOGY_FAILURE.set(null);
+ }
+
+ @Override
+ public CompletableFuture<Void> setTopology(String groupId, int topologyEpoch, StreamsGroupTopologyDescription description) {
+ switch (SET_TOPOLOGY_FAILURE_MODE.getAndSet(SetTopologyFailureMode.NONE)) {
+ case PERMANENT:
+ return CompletableFuture.failedFuture(
+ new StreamsTopologyDescriptionPermanentFailureException("topology rejected by test plugin"));
+ case TRANSIENT:
+ return CompletableFuture.failedFuture(
+ new StreamsTopologyDescriptionTransientFailureException("backend offline"));
+ default:
+ return super.setTopology(groupId, topologyEpoch, description);
+ }
+ }
+
+ @Override
+ public CompletableFuture<Void> deleteTopology(String groupId) {
+ RuntimeException failure = DELETE_TOPOLOGY_FAILURE.get();
+ if (failure != null) {
+ return CompletableFuture.failedFuture(failure);
+ }
+ return super.deleteTopology(groupId);
+ }
+}
diff --git a/core/src/test/scala/unit/kafka/server/StreamsGroupTopologyDescriptionRequestTest.scala b/core/src/test/scala/unit/kafka/server/StreamsGroupTopologyDescriptionRequestTest.scala
index 8352339f2ce..454bd061382 100644
--- a/core/src/test/scala/unit/kafka/server/StreamsGroupTopologyDescriptionRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/StreamsGroupTopologyDescriptionRequestTest.scala
@@ -18,13 +18,14 @@ 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.message.{DeleteGroupsRequestData, StreamsGroupHeartbeatRequestData, StreamsGroupTopologyDescriptionUpdateRequestData}
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
-import org.apache.kafka.common.requests.StreamsGroupDescribeResponse
+import org.apache.kafka.common.requests.{DeleteGroupsRequest, DeleteGroupsResponse, 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 org.junit.jupiter.api.Timeout
import scala.jdk.CollectionConverters._
@@ -307,7 +308,7 @@ class StreamsGroupTopologyDescriptionRequestTest(cluster: ClusterInstance) exten
deleteGroups(
groupIds = List(groupId),
expectedErrors = List(Errors.NONE),
- version = ApiKeys.DELETE_GROUPS.latestVersion(isUnstableApiEnabled)
+ version = ApiKeys.DELETE_GROUPS.latestVersion()
)
// The group is gone.
@@ -321,6 +322,301 @@ class StreamsGroupTopologyDescriptionRequestTest(cluster: ClusterInstance) exten
}
}
+ @ClusterTest(serverProperties = Array(
+ new ClusterConfigProperty(
+ key = GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG,
+ value = "kafka.server.FailingTopologyDescriptionPlugin")
+ ))
+ def testStreamsGroupTopologyDescriptionUpdatePermanentFailureRatchetsFailedEpoch(): Unit = {
+ val admin = cluster.admin()
+ val groupId = "test-group"
+ val memberId = "test-member"
+ val topicName = "test-topic"
+
+ try {
+ FailingTopologyDescriptionPlugin.reset()
+ 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
+ )
+
+ val memberEpoch = joinAndAwaitTopologyDescriptionSolicited(groupId, memberId, topicName)
+
+ FailingTopologyDescriptionPlugin.failNextSetTopology(FailingTopologyDescriptionPlugin.SetTopologyFailureMode.PERMANENT)
+
+ val updateResponse = streamsGroupTopologyDescriptionUpdate(
+ groupId = groupId,
+ memberId = memberId,
+ topologyEpoch = topologyEpoch,
+ topologyDescription = createTopologyDescription(topicName)
+ )
+ assertEquals(Errors.STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED.code(), updateResponse.errorCode())
+ assertEquals("topology rejected by test plugin", updateResponse.errorMessage())
+
+ // A permanent failure ratchets the group's failed topology epoch: the broker must not
+ // re-solicit another push at the same epoch.
+ val heartbeatAfterFailure = streamsGroupHeartbeat(
+ groupId = groupId,
+ memberId = memberId,
+ memberEpoch = memberEpoch,
+ rebalanceTimeoutMs = 1000,
+ activeTasks = List.empty,
+ standbyTasks = List.empty,
+ warmupTasks = List.empty
+ )
+ assertFalse(heartbeatAfterFailure.topologyDescriptionRequired(),
+ "Broker must not re-solicit a topology description push once the epoch is marked permanently failed.")
+
+ // Restarting the broker replaces its StreamsGroupTopologyDescriptionBackoff with a fresh,
+ // empty one, while the group's failed-topology-epoch record survives in the persisted
+ // coordinator log. If the heartbeat below is still suppressed, that can only be the
+ // persisted ratchet: neither a leftover back-off window from the original solicitation
+ // nor a transient-failure back-off window would survive the restart.
+ val brokerId = cluster.brokerIds().iterator().next()
+ cluster.restartBroker(brokerId, java.util.Collections.emptyMap())
+ cluster.waitForReadyBrokers()
+
+ val heartbeatAfterRestart = streamsGroupHeartbeat(
+ groupId = groupId,
+ memberId = memberId,
+ memberEpoch = memberEpoch,
+ rebalanceTimeoutMs = 1000,
+ activeTasks = List.empty,
+ standbyTasks = List.empty,
+ warmupTasks = List.empty
+ )
+ assertFalse(heartbeatAfterRestart.topologyDescriptionRequired(),
+ "Broker must still ratchet the failed topology epoch after a restart clears in-memory back-off state.")
+
+ // Nothing was ever stored for this epoch: the permanent failure leaves the group's
+ // stored-topology-epoch at its UNCERTAIN sentinel, which Describe reports as NOT_STORED.
+ val describedGroup = streamsGroupDescribe(
+ groupIds = List(groupId),
+ includeTopologyDescription = true
+ ).head
+ assertEquals(StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_NOT_STORED, describedGroup.topologyDescriptionStatus())
+ } finally {
+ FailingTopologyDescriptionPlugin.reset()
+ admin.close()
+ }
+ }
+
+ @Timeout(180)
+ @ClusterTest(serverProperties = Array(
+ new ClusterConfigProperty(
+ key = GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG,
+ value = "kafka.server.FailingTopologyDescriptionPlugin"),
+ // The test below goes 40s without heartbeating; the default 45s session timeout would
+ // leave only ~5s of margin before the member is reaped for inactivity.
+ new ClusterConfigProperty(key = GroupCoordinatorConfig.STREAMS_GROUP_SESSION_TIMEOUT_MS_CONFIG, value = "120000"),
+ new ClusterConfigProperty(key = GroupCoordinatorConfig.STREAMS_GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG, value = "120000")
+ ))
+ def testStreamsGroupTopologyDescriptionUpdateTransientFailureArmsBackOff(): Unit = {
+ val admin = cluster.admin()
+ val groupId = "test-group"
+ val memberId = "test-member"
+ val topicName = "test-topic"
+
+ try {
+ FailingTopologyDescriptionPlugin.reset()
+ 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. This
+ // arms the first back-off window: 30s * 2^0, +/-20% jitter (StreamsGroupTopologyDescriptionBackoff),
+ // i.e. 24-36s.
+ var memberEpoch = joinAndAwaitTopologyDescriptionSolicited(groupId, memberId, topicName)
+
+ // Let that window fully expire without heartbeating, so nothing but the upcoming
+ // transient failure can be responsible for the suppression asserted below. The back-off
+ // uses the broker's real clock, with no test seam to inject a mock one here, so this
+ // waits out real time rather than simulating it.
+ Thread.sleep(40000)
+
+ FailingTopologyDescriptionPlugin.failNextSetTopology(FailingTopologyDescriptionPlugin.SetTopologyFailureMode.TRANSIENT)
+
+ val updateResponse = streamsGroupTopologyDescriptionUpdate(
+ groupId = groupId,
+ memberId = memberId,
+ topologyEpoch = topologyEpoch,
+ topologyDescription = createTopologyDescription(topicName)
+ )
+ assertEquals(Errors.STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED.code(), updateResponse.errorCode())
+ assertEquals("backend offline", updateResponse.errorMessage())
+
+ // The first window already lapsed, so this suppression can only come from the back-off
+ // the transient failure itself just armed (30s * 2^1, +/-20% jitter, i.e. 48-72s).
+ val heartbeatAfterFailure = streamsGroupHeartbeat(
+ groupId = groupId,
+ memberId = memberId,
+ memberEpoch = memberEpoch,
+ rebalanceTimeoutMs = 1000,
+ activeTasks = List.empty,
+ standbyTasks = List.empty,
+ warmupTasks = List.empty
+ )
+ assertFalse(heartbeatAfterFailure.topologyDescriptionRequired(),
+ "Broker must not re-solicit a topology description push immediately after a transient failure; back-off must be armed.")
+
+ // Unlike a permanent failure, this back-off lapses on its own: the broker eventually
+ // re-solicits at the same topology epoch, which is what distinguishes transient handling
+ // from ratcheting the epoch as permanently failed.
+ TestUtils.waitUntilTrue(() => {
+ val response = streamsGroupHeartbeat(
+ groupId = groupId,
+ memberId = memberId,
+ memberEpoch = memberEpoch,
+ rebalanceTimeoutMs = 1000,
+ activeTasks = List.empty,
+ standbyTasks = List.empty,
+ warmupTasks = List.empty
+ )
+ memberEpoch = response.memberEpoch()
+ response.errorCode == Errors.NONE.code() && response.topologyDescriptionRequired()
+ }, "Broker did not re-solicit a topology description push after the transient back-off lapsed.", waitTimeMs = 90000, pause = 2000L)
+ } finally {
+ FailingTopologyDescriptionPlugin.reset()
+ admin.close()
+ }
+ }
+
+ @ClusterTest(serverProperties = Array(
+ new ClusterConfigProperty(
+ key = GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG,
+ value = "kafka.server.FailingTopologyDescriptionPlugin")
+ ))
+ def testDeleteGroupsPluginFailureBlocksTombstoneUntilRecovered(): Unit = {
+ val admin = cluster.admin()
+ val groupId = "test-group"
+ val memberId = "test-member"
+ val topicName = "test-topic"
+
+ try {
+ FailingTopologyDescriptionPlugin.reset()
+ 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 successfully.
+ 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()}")
+
+ // 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())
+
+ FailingTopologyDescriptionPlugin.failDeleteTopologyWith(new RuntimeException("plugin offline"))
+
+ val deleteVersion = ApiKeys.DELETE_GROUPS.latestVersion()
+ val failedDeleteResponse = deleteGroupsRaw(List(groupId), deleteVersion)
+ val failedResult = failedDeleteResponse.data.results.find(groupId)
+ assertNotNull(failedResult)
+ assertEquals(Errors.GROUP_DELETION_FAILED.code(), failedResult.errorCode())
+
+ // The group survives a failed delete: Describe must still find it.
+ val describedAfterFailedDelete = streamsGroupDescribe(
+ groupIds = List(groupId),
+ includeTopologyDescription = true
+ ).head
+ assertEquals(Errors.NONE.code(), describedAfterFailedDelete.errorCode())
+
+ // Restore the plugin: a retried DeleteGroups must succeed and tombstone the group.
+ FailingTopologyDescriptionPlugin.reset()
+ deleteGroups(
+ groupIds = List(groupId),
+ expectedErrors = List(Errors.NONE),
+ version = deleteVersion
+ )
+
+ val describedAfterDelete = streamsGroupDescribe(
+ groupIds = List(groupId),
+ includeTopologyDescription = true
+ ).head
+ assertEquals(Errors.GROUP_ID_NOT_FOUND.code(), describedAfterDelete.errorCode())
+ } finally {
+ FailingTopologyDescriptionPlugin.reset()
+ admin.close()
+ }
+ }
+
+ private def joinAndAwaitTopologyDescriptionSolicited(groupId: String, memberId: String, topicName: String): Int = {
+ 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)
+ )
+ memberEpoch = response.memberEpoch()
+ response.errorCode == Errors.NONE.code() && response.topologyDescriptionRequired()
+ }, "Broker did not solicit a topology description push within the timeout period.")
+ memberEpoch
+ }
+
+ private def deleteGroupsRaw(groupIds: List[String], version: Short): DeleteGroupsResponse = {
+ val deleteGroupsRequest = new DeleteGroupsRequest.Builder(
+ new DeleteGroupsRequestData().setGroupsNames(groupIds.asJava)
+ ).build(version)
+ connectAndReceive[DeleteGroupsResponse](deleteGroupsRequest)
+ }
+
private def createMockTopology(topicName: String): StreamsGroupHeartbeatRequestData.Topology = {
new StreamsGroupHeartbeatRequestData.Topology()
.setEpoch(topologyEpoch)