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/09 07:21:24 UTC
(kafka) branch trunk updated: MINOR: Add broker integration tests for classic JoinGroup topology cleanup (#23747)
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 3c1b3f2db34 MINOR: Add broker integration tests for classic JoinGroup topology cleanup (#23747)
3c1b3f2db34 is described below
commit 3c1b3f2db34b2191c00bd97d3de92c4945668ab6
Author: Lucas Brutschy <lb...@confluent.io>
AuthorDate: Fri Oct 9 09:21:15 2026 +0200
MINOR: Add broker integration tests for classic JoinGroup topology cleanup (#23747)
Converting an empty streams group to a classic group requires the broker
to delete the group's stored topology description from the plugin first.
This path was only unit-tested against mocks; no test sent a real
classic JoinGroupRequest to a running broker, and the failed-join and
periodic-cleanup halves were never chained together.
This adds three single-broker integration tests to
StreamsGroupTopologyDescriptionRequestTest. The first joins a classic
member to an empty streams group with a stored topology and checks that
the broker calls plugin.deleteTopology and converts the group. The
second makes deleteTopology fail and checks that the join is rejected
with the retriable REBALANCE_IN_PROGRESS while the group stays a streams
group. The third chains the failure with the periodic cleanup cycle:
after the plugin recovers, the cycle clears the stored topology without
any client involvement, and a later classic join converts the group.
FailingTopologyDescriptionPlugin gains a per-group counter of
deleteTopology invocations, so the tests can assert that the broker
actually reached the plugin.
Reviewers: TengYao Chi <fr...@apache.org>
---
.../server/FailingTopologyDescriptionPlugin.java | 15 ++
...treamsGroupTopologyDescriptionRequestTest.scala | 162 ++++++++++++++++++++-
2 files changed, 176 insertions(+), 1 deletion(-)
diff --git a/core/src/test/java/kafka/server/FailingTopologyDescriptionPlugin.java b/core/src/test/java/kafka/server/FailingTopologyDescriptionPlugin.java
index 0fae2199176..61b459bde0b 100644
--- a/core/src/test/java/kafka/server/FailingTopologyDescriptionPlugin.java
+++ b/core/src/test/java/kafka/server/FailingTopologyDescriptionPlugin.java
@@ -22,6 +22,8 @@ import org.apache.kafka.coordinator.group.api.streams.StreamsTopologyDescription
import org.apache.kafka.server.streams.InMemoryTopologyDescriptionPlugin;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
/**
@@ -47,6 +49,8 @@ public class FailingTopologyDescriptionPlugin extends InMemoryTopologyDescriptio
new AtomicReference<>(SetTopologyFailureMode.NONE);
private static final AtomicReference<RuntimeException> DELETE_TOPOLOGY_FAILURE =
new AtomicReference<>(null);
+ private static final ConcurrentHashMap<String, AtomicInteger> DELETE_TOPOLOGY_ATTEMPTS =
+ new ConcurrentHashMap<>();
public static void failNextSetTopology(SetTopologyFailureMode mode) {
SET_TOPOLOGY_FAILURE_MODE.set(mode);
@@ -59,6 +63,16 @@ public class FailingTopologyDescriptionPlugin extends InMemoryTopologyDescriptio
public static void reset() {
SET_TOPOLOGY_FAILURE_MODE.set(SetTopologyFailureMode.NONE);
DELETE_TOPOLOGY_FAILURE.set(null);
+ DELETE_TOPOLOGY_ATTEMPTS.clear();
+ }
+
+ /**
+ * @return how many times {@code deleteTopology} was invoked for the group since the last
+ * {@link #reset()}, whether or not the call was configured to fail.
+ */
+ public static int deleteTopologyAttempts(String groupId) {
+ AtomicInteger attempts = DELETE_TOPOLOGY_ATTEMPTS.get(groupId);
+ return attempts == null ? 0 : attempts.get();
}
@Override
@@ -77,6 +91,7 @@ public class FailingTopologyDescriptionPlugin extends InMemoryTopologyDescriptio
@Override
public CompletableFuture<Void> deleteTopology(String groupId) {
+ DELETE_TOPOLOGY_ATTEMPTS.computeIfAbsent(groupId, k -> new AtomicInteger()).incrementAndGet();
RuntimeException failure = DELETE_TOPOLOGY_FAILURE.get();
if (failure != null) {
return CompletableFuture.failedFuture(failure);
diff --git a/core/src/test/scala/unit/kafka/server/StreamsGroupTopologyDescriptionRequestTest.scala b/core/src/test/scala/unit/kafka/server/StreamsGroupTopologyDescriptionRequestTest.scala
index 454bd061382..e8b4ed88b11 100644
--- a/core/src/test/scala/unit/kafka/server/StreamsGroupTopologyDescriptionRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/StreamsGroupTopologyDescriptionRequestTest.scala
@@ -17,8 +17,9 @@
package kafka.server
import kafka.utils.TestUtils
+import org.apache.kafka.clients.admin.Admin
import org.apache.kafka.common.errors.UnsupportedVersionException
-import org.apache.kafka.common.message.{DeleteGroupsRequestData, StreamsGroupHeartbeatRequestData, StreamsGroupTopologyDescriptionUpdateRequestData}
+import org.apache.kafka.common.message.{DeleteGroupsRequestData, JoinGroupResponseData, StreamsGroupHeartbeatRequestData, StreamsGroupTopologyDescriptionUpdateRequestData}
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.common.requests.{DeleteGroupsRequest, DeleteGroupsResponse, StreamsGroupDescribeResponse}
import org.apache.kafka.common.test.ClusterInstance
@@ -592,6 +593,165 @@ class StreamsGroupTopologyDescriptionRequestTest(cluster: ClusterInstance) exten
}
}
+ @ClusterTest(serverProperties = Array(
+ new ClusterConfigProperty(
+ key = GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG,
+ value = "kafka.server.FailingTopologyDescriptionPlugin"),
+ // Keeps the periodic cleanup cycle from also deleting the topology while the test runs.
+ new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_RETENTION_CHECK_INTERVAL_MS_CONFIG, value = "3600000")
+ ))
+ def testClassicJoinDeletesStoredTopologyAndConvertsEmptyStreamsGroup(): Unit = {
+ val admin = cluster.admin()
+ val groupId = "test-group"
+ val topicName = "test-topic"
+
+ try {
+ FailingTopologyDescriptionPlugin.reset()
+ createEmptyStreamsGroupWithStoredTopology(admin, groupId, topicName)
+
+ assertEquals(Errors.NONE.code(), classicJoin(groupId).errorCode())
+ assertEquals(1, FailingTopologyDescriptionPlugin.deleteTopologyAttempts(groupId))
+ assertEquals(Errors.GROUP_ID_NOT_FOUND.code(), streamsGroupDescribe(List(groupId)).head.errorCode())
+ val describedClassicGroup = describeGroups(List(groupId)).head
+ assertEquals(Errors.NONE.code(), describedClassicGroup.errorCode())
+ assertEquals("consumer", describedClassicGroup.protocolType())
+ } finally {
+ FailingTopologyDescriptionPlugin.reset()
+ admin.close()
+ }
+ }
+
+ @ClusterTest(serverProperties = Array(
+ new ClusterConfigProperty(
+ key = GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG,
+ value = "kafka.server.FailingTopologyDescriptionPlugin")
+ ))
+ def testClassicJoinIsRejectedRetriablyWhenPluginDeleteFails(): Unit = {
+ val admin = cluster.admin()
+ val groupId = "test-group"
+ val topicName = "test-topic"
+
+ try {
+ FailingTopologyDescriptionPlugin.reset()
+ createEmptyStreamsGroupWithStoredTopology(admin, groupId, topicName)
+
+ FailingTopologyDescriptionPlugin.failDeleteTopologyWith(new RuntimeException("plugin offline"))
+ assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), classicJoin(groupId).errorCode())
+ assertEquals(1, FailingTopologyDescriptionPlugin.deleteTopologyAttempts(groupId))
+ assertStillStreamsGroup(groupId)
+ } finally {
+ FailingTopologyDescriptionPlugin.reset()
+ admin.close()
+ }
+ }
+
+ @Timeout(120)
+ @ClusterTest(serverProperties = Array(
+ new ClusterConfigProperty(
+ key = GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG,
+ value = "kafka.server.FailingTopologyDescriptionPlugin"),
+ // Drives the periodic topology cleanup cycle, which runs on the offsets retention check interval.
+ new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_RETENTION_CHECK_INTERVAL_MS_CONFIG, value = "500")
+ ))
+ def testPeriodicCleanupClearsTopologyAfterFailedClassicJoin(): Unit = {
+ val admin = cluster.admin()
+ val groupId = "test-group"
+ val topicName = "test-topic"
+
+ try {
+ FailingTopologyDescriptionPlugin.reset()
+ // Fail deletes before the group becomes empty: once it is, the cleanup cycle (500 ms) may
+ // run at any moment and would otherwise delete the topology before the join is attempted.
+ FailingTopologyDescriptionPlugin.failDeleteTopologyWith(new RuntimeException("plugin offline"))
+ createEmptyStreamsGroupWithStoredTopology(admin, groupId, topicName)
+
+ // The plugin cannot delete: the classic join is rejected and the group stays a streams group.
+ assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), classicJoin(groupId).errorCode())
+ assertStillStreamsGroup(groupId)
+
+ // The plugin recovers. The failed join armed a back-off (at least 24s) that stops further
+ // joins from calling the plugin, and only a successful cleanup cycle clears it. So a join
+ // that gets past REBALANCE_IN_PROGRESS shows the periodic cycle emptied the plugin.
+ FailingTopologyDescriptionPlugin.failDeleteTopologyWith(null)
+ var joinResponse: JoinGroupResponseData = null
+ TestUtils.waitUntilTrue(() => {
+ joinResponse = sendJoinRequest(groupId = groupId)
+ joinResponse.errorCode() != Errors.REBALANCE_IN_PROGRESS.code()
+ }, "Classic join was still rejected after the plugin recovered.")
+
+ // The group converts once the member re-joins with the member id the broker assigned.
+ assertEquals(Errors.MEMBER_ID_REQUIRED.code(), joinResponse.errorCode())
+ assertEquals(Errors.NONE.code(), sendJoinRequest(groupId = groupId, memberId = joinResponse.memberId()).errorCode())
+ assertEquals(Errors.GROUP_ID_NOT_FOUND.code(), streamsGroupDescribe(List(groupId)).head.errorCode())
+ val describedClassicGroup = describeGroups(List(groupId)).head
+ assertEquals(Errors.NONE.code(), describedClassicGroup.errorCode())
+ assertEquals("consumer", describedClassicGroup.protocolType())
+ } finally {
+ FailingTopologyDescriptionPlugin.reset()
+ admin.close()
+ }
+ }
+
+ /**
+ * Joins as a new classic member. The topology cleanup runs on the first join; the group only
+ * converts to classic once the member re-joins with the member id the broker assigns
+ * (MEMBER_ID_REQUIRED), so this follows that second step.
+ */
+ private def classicJoin(groupId: String): JoinGroupResponseData = {
+ val firstResponse = sendJoinRequest(groupId = groupId)
+ if (firstResponse.errorCode() != Errors.MEMBER_ID_REQUIRED.code()) {
+ firstResponse
+ } else {
+ sendJoinRequest(groupId = groupId, memberId = firstResponse.memberId())
+ }
+ }
+
+ /**
+ * Creates a streams group whose only member has left, leaving its topology description
+ * stored in the plugin with no members remaining — the state in which a classic join
+ * must delete the topology before converting the group.
+ */
+ private def createEmptyStreamsGroupWithStoredTopology(admin: Admin, groupId: String, topicName: String): Unit = {
+ val memberId = "test-member"
+ 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
+ )
+
+ joinAndAwaitTopologyDescriptionSolicited(groupId, memberId, topicName)
+ val updateResponse = streamsGroupTopologyDescriptionUpdate(
+ groupId = groupId,
+ memberId = memberId,
+ topologyEpoch = topologyEpoch,
+ topologyDescription = createTopologyDescription(topicName)
+ )
+ assertEquals(Errors.NONE.code(), updateResponse.errorCode(), s"Unexpected error: ${updateResponse.errorMessage()}")
+
+ 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())
+ }
+
+ private def assertStillStreamsGroup(groupId: String): Unit = {
+ val describedGroup = streamsGroupDescribe(List(groupId)).head
+ assertEquals(Errors.NONE.code(), describedGroup.errorCode())
+ }
+
private def joinAndAwaitTopologyDescriptionSolicited(groupId: String, memberId: String, topicName: String): Int = {
var memberEpoch = 0
TestUtils.waitUntilTrue(() => {