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 5b70498523d KAFKA-20623: DeleteGroups extension for streams group
topology description plugin (3/3) (#22554)
5b70498523d is described below
commit 5b70498523d95932383969d921d71d324e91bbd1
Author: TengYao Chi <[email protected]>
AuthorDate: Thu Jun 18 07:36:09 2026 +0100
KAFKA-20623: DeleteGroups extension for streams group topology description
plugin (3/3) (#22554)
JIRA: KAFKA-20623
This is a part of KIP-1331
**This PR shouldn't be merge before #22552**
Final of three stacked sub-tasks. The `deleteGroups` flow now calls
`plugin.deleteTopology` for every streams group with a stored topology
description before tombstoning. Plugin failure surfaces as
`GROUP_DELETION_FAILED` (downgraded to `UNKNOWN_SERVER_ERROR` for
DeleteGroups v2) and the group is held back from tombstoning; mixed
batches are honoured per-group.
### Pre-tombstone deletion hook on the manager
`TopologyDescriptionManager.deleteBeforeGroupDelete(topicPartition,
groupIds)` calls
`plugin.deleteTopology` in parallel for every group id in the batch that
is a streams
group with a non-default `StoredDescriptionTopologyEpoch`. The
eligibility filter is
resolved through `scheduleReadOperation` so it sees the same shard
snapshot used by
the subsequent tombstone write.
Returns a per-group failure map keyed by group id. Groups absent from
the map either had no plugin state to clean up or the plugin call
succeeded; groups present in the map carry an `ApiError` describing the
failure. A read failure (NOT_COORDINATOR etc.) on the eligibility query
is treated the same as a plugin failure: every retained group in the
batch is held back from tombstoning so the caller can retry.
The manager re-gains the `LogContext` constructor parameter so it can
emit a broker-side `WARN` for each plugin failure alongside the
per-group response error.
### Service-layer chain extension
`GroupCoordinatorService.deleteGroups` now splices the deletion hook
between the
share-group pre-step and the underlying `delete-groups` write:
1. `deleteShareGroups` filters share-group failures (unchanged).
2. `topologyDescriptionManager.deleteBeforeGroupDelete` runs for the
retained ids.
3. `filterStreamsTopologyErrors` moves per-group plugin failures into
the response
collection and returns the subset that should still proceed to
tombstoning.
4. `handleDeleteGroups` writes tombstones for the retained subset.
If every retained group fails the plugin call, the underlying
`delete-groups` write is skipped entirely. Mixed batches are honoured
per-group — the successful ids reach the tombstone write, the failed
ids carry `GROUP_DELETION_FAILED` in the response, and an idempotent
retry of `DeleteGroups` converges once the plugin recovers.
### v2 downgrade
Per KIP-1331, `GROUP_DELETION_FAILED` and the per-group `ErrorMessage`
field were
introduced at `DeleteGroups` v3. For older clients the broker downgrades
the error
code to `UNKNOWN_SERVER_ERROR` with no message — matching the convention
used by
KIP-1043 for new error codes that pre-existing request versions cannot
interpret.
Errors that predate this KIP (e.g. `NOT_COORDINATOR` surfaced from the
runtime)
pass through unchanged on all versions.
### Coordinator shard + GMM
`GroupMetadataManager.streamsGroupsWithStoredTopologyDescription(groupIds,
committedOffset)`
returns the subset of the input that is (a) a streams group on this
shard and
(b) carries a non-default `StoredDescriptionTopologyEpoch`. Non-existent
groups and
non-streams groups are silently skipped — they have no plugin state to
clean up.
`GroupCoordinatorShard` exposes the method to the runtime read-operation
scheduler.
Reviewers: Lucas Brutschy <[email protected]>
---
core/src/main/scala/kafka/server/KafkaApis.scala | 11 +
.../scala/unit/kafka/server/KafkaApisTest.scala | 66 ++++++
.../coordinator/group/GroupCoordinatorService.java | 141 +++++++++++-
.../coordinator/group/GroupCoordinatorShard.java | 13 ++
.../coordinator/group/GroupMetadataManager.java | 30 +++
.../StreamsGroupTopologyDescriptionManager.java | 61 +++++
...pCoordinatorServiceTopologyDescriptionTest.java | 255 +++++++++++++++++++++
7 files changed, 573 insertions(+), 4 deletions(-)
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala
b/core/src/main/scala/kafka/server/KafkaApis.scala
index 959e7c76dc3..a3058af66c4 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -1456,6 +1456,17 @@ class KafkaApis(val requestChannel: RequestChannel,
.setErrorCode(error.code))
}
} else {
+ // GROUP_DELETION_FAILED was introduced for DeleteGroups v3
(KIP-1331). Older
+ // clients cannot interpret the new code, so downgrade it to
UNKNOWN_SERVER_ERROR.
+ // ErrorMessage is "versions": "3+", "ignorable": true and is stripped
at the
+ // serialization layer for v<3, so only the error code needs gating
here.
+ if (request.context.apiVersion < 3) {
+ results.forEach { result =>
+ if (result.errorCode == Errors.GROUP_DELETION_FAILED.code) {
+ result.setErrorCode(Errors.UNKNOWN_SERVER_ERROR.code)
+ }
+ }
+ }
response.setResults(results)
}
diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
index 701f2db0df0..e9a196118f2 100644
--- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
@@ -3633,6 +3633,72 @@ class KafkaApisTest extends Logging {
assertEquals(expectedDeleteGroupsResponse, response.data)
}
+ @Test
+ def testHandleDeleteGroupsDowngradesGroupDeletionFailedForOldClients(): Unit
= {
+ // GROUP_DELETION_FAILED was introduced for DeleteGroups v3 (KIP-1331).
For v<3 clients
+ // KafkaApis must downgrade it to UNKNOWN_SERVER_ERROR so the client can
interpret the
+ // code; the ErrorMessage field is "versions": "3+", "ignorable": true and
is stripped
+ // by the serialization layer for v<3, so this layer does not need to
clear it.
+ val deleteGroupsRequest = new
DeleteGroupsRequestData().setGroupsNames(util.List.of("group-1"))
+ val requestChannelRequest = buildRequest(new
DeleteGroupsRequest.Builder(deleteGroupsRequest).build(2))
+
+ val future = new
CompletableFuture[DeleteGroupsResponseData.DeletableGroupResultCollection]()
+ when(groupCoordinator.deleteGroups(
+ requestChannelRequest.context,
+ util.List.of("group-1"),
+ RequestLocal.noCaching.bufferSupplier
+ )).thenReturn(future)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleDeleteGroupsRequest(
+ requestChannelRequest,
+ RequestLocal.noCaching
+ )
+
+ future.complete(new
DeleteGroupsResponseData.DeletableGroupResultCollection(util.List.of(
+ new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("group-1")
+ .setErrorCode(Errors.GROUP_DELETION_FAILED.code)
+ .setErrorMessage("plugin offline")
+ ).iterator))
+
+ val response =
verifyNoThrottling[DeleteGroupsResponse](requestChannelRequest)
+ val result = response.data.results.find("group-1")
+ assertNotNull(result)
+ assertEquals(Errors.UNKNOWN_SERVER_ERROR.code, result.errorCode)
+ }
+
+ @Test
+ def testHandleDeleteGroupsKeepsGroupDeletionFailedForV3Clients(): Unit = {
+ // v3+ clients understand GROUP_DELETION_FAILED, so no downgrade.
+ val deleteGroupsRequest = new
DeleteGroupsRequestData().setGroupsNames(util.List.of("group-1"))
+ val requestChannelRequest = buildRequest(new
DeleteGroupsRequest.Builder(deleteGroupsRequest).build(3))
+
+ val future = new
CompletableFuture[DeleteGroupsResponseData.DeletableGroupResultCollection]()
+ when(groupCoordinator.deleteGroups(
+ requestChannelRequest.context,
+ util.List.of("group-1"),
+ RequestLocal.noCaching.bufferSupplier
+ )).thenReturn(future)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleDeleteGroupsRequest(
+ requestChannelRequest,
+ RequestLocal.noCaching
+ )
+
+ future.complete(new
DeleteGroupsResponseData.DeletableGroupResultCollection(util.List.of(
+ new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("group-1")
+ .setErrorCode(Errors.GROUP_DELETION_FAILED.code)
+ .setErrorMessage("plugin offline")
+ ).iterator))
+
+ val response =
verifyNoThrottling[DeleteGroupsResponse](requestChannelRequest)
+ val result = response.data.results.find("group-1")
+ assertNotNull(result)
+ assertEquals(Errors.GROUP_DELETION_FAILED.code, result.errorCode)
+ assertEquals("plugin offline", result.errorMessage)
+ }
+
@Test
def testHandleDeleteGroupsFutureFailed(): Unit = {
val deleteGroupsRequest = new
DeleteGroupsRequestData().setGroupsNames(util.List.of(
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
index 1f78c0b401f..2708b7b261b 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
@@ -20,6 +20,8 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.compress.Compression;
import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.errors.CoordinatorLoadInProgressException;
+import org.apache.kafka.common.errors.CoordinatorNotAvailableException;
import org.apache.kafka.common.errors.GroupIdNotFoundException;
import org.apache.kafka.common.errors.InvalidRequestException;
import org.apache.kafka.common.errors.NotCoordinatorException;
@@ -795,12 +797,24 @@ public class GroupCoordinatorService implements
GroupCoordinator {
if (throwable == null) {
return new SetTopologyOutcome(responseOnSuccess,
BackoffAction.CLEAR, null);
}
+ Throwable cause = Errors.maybeUnwrapException(throwable);
// The group was deleted between the plugin call and the bookkeeping
write — the
// push already took effect at the plugin and no live group remains to
throttle,
// so drop the orphaned back-off entry instead of arming one nobody
will clear.
- if (Errors.maybeUnwrapException(throwable) instanceof
GroupIdNotFoundException) {
+ if (cause instanceof GroupIdNotFoundException) {
return new SetTopologyOutcome(null, BackoffAction.CLEAR_GROUP,
throwable);
}
+ // This broker stopped being the coordinator between the plugin call
and the
+ // bookkeeping write. The client will retry against the new
coordinator, which
+ // holds no back-off entry of its own; arming a broker-wide entry here
would
+ // leak until expiry and could suppress a legitimate solicitation if
the group
+ // ever migrates back. The retry that lands on the new coordinator is
what
+ // re-establishes convergence, not a back-off window on the broker
that bowed out.
+ if (cause instanceof NotCoordinatorException
+ || cause instanceof CoordinatorLoadInProgressException
+ || cause instanceof CoordinatorNotAvailableException) {
+ return new SetTopologyOutcome(null, BackoffAction.NOOP, throwable);
+ }
return new SetTopologyOutcome(null, BackoffAction.ARM, throwable);
}
@@ -1627,9 +1641,42 @@ public class GroupCoordinatorService implements
GroupCoordinator {
return
CompletableFuture.completedFuture(deletableGroupResults);
}
- return handleDeleteGroups(context, topicPartition,
retainedGroupIds)
- .whenComplete((resp, __) -> resp.forEach(result ->
deletableGroupResults.add(result.duplicate())))
- .thenApply(__ -> deletableGroupResults);
+ return runPrePluginDelete(topicPartition, retainedGroupIds)
+ .thenCompose(streamsErrMap -> {
+ List<String> afterStreams =
filterStreamsTopologyErrors(
+ streamsErrMap, retainedGroupIds,
deletableGroupResults);
+ if (afterStreams.isEmpty()) {
+ return
CompletableFuture.completedFuture(deletableGroupResults);
+ }
+ return handleDeleteGroups(context, topicPartition,
afterStreams)
+ .whenComplete((resp, __) -> resp.forEach(result ->
deletableGroupResults.add(result.duplicate())))
+ .thenApply(__ -> deletableGroupResults);
+ })
+ .exceptionally(exception -> {
+ // Defensive net for any uncaught synchronous throw in
the
+ // post-runPrePluginDelete stage. Without this, the
exception would
+ // propagate through FutureUtils.combineFutures.join()
and fail the
+ // whole cross-partition DeleteGroups response —
including groups on
+ // other partitions that already succeeded. Runtime
read failures
+ // inside runPrePluginDelete are absorbed there, so
they never reach
+ // this branch; what we are catching here is the
synchronous stages
+ // (filterStreamsTopologyErrors etc.). Fold the
exception into
+ // per-group failures for any retainedGroupIds not yet
recorded.
+ ApiError apiError = ApiError.fromThrowable(exception);
+ Set<String> recorded = new HashSet<>();
+ deletableGroupResults.forEach(result ->
recorded.add(result.groupId()));
+ for (String groupId : retainedGroupIds) {
+ if (!recorded.contains(groupId)) {
+ deletableGroupResults.add(
+ new
DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId(groupId)
+ .setErrorCode(apiError.error().code())
+ .setErrorMessage(apiError.message())
+ );
+ }
+ }
+ return deletableGroupResults;
+ });
});
// deleteShareGroups has its own exceptionally block, so we don't
need one here.
@@ -1688,6 +1735,92 @@ public class GroupCoordinatorService implements
GroupCoordinator {
return groupSet.stream().toList();
}
+ /**
+ * Drive the topology-description plugin's pre-delete flow: identify which
of the
+ * supplied group ids carry a stored topology description, call
+ * {@code plugin.deleteTopology} for each, and drop the corresponding
back-off entries.
+ *
+ * <p>Short-circuits with an empty failure map when no plugin is
configured, so a
+ * broker with the feature disabled does not hit the runtime for a
per-partition read.
+ * The chain mirrors the structure of {@code
streamsGroupTopologyDescriptionUpdate}:
+ * the manager exposes pure plugin invocation ({@code
invokeDeleteTopologies}) plus
+ * a back-off mutation ({@code clearBackoffGroup}), and this service
helper assembles
+ * the runtime read, the plugin call, and the back-off cleanup into one
future.
+ *
+ * <p>Runtime read failures (e.g. {@code NOT_COORDINATOR}) are folded back
into the
+ * same per-group failure map so the caller can report them uniformly; we
deliberately
+ * pass through the more specific runtime error rather than collapsing
everything to
+ * {@code GROUP_DELETION_FAILED}.
+ */
+ private CompletableFuture<Map<String, ApiError>> runPrePluginDelete(
+ TopicPartition topicPartition,
+ List<String> groupIds
+ ) {
+ if (!streamsGroupTopologyDescriptionManager.isPluginConfigured()) {
+ return CompletableFuture.completedFuture(Map.of());
+ }
+ return runtime.scheduleReadOperation(
+ "streams-group-topology-pre-delete",
+ topicPartition,
+ (coordinator, lastCommittedOffset) ->
+
coordinator.streamsGroupsWithStoredTopologyDescription(groupIds,
lastCommittedOffset))
+ .thenCompose(groupsWithStored ->
+
streamsGroupTopologyDescriptionManager.invokeDeleteTopologies(groupsWithStored)
+ .thenApply(failures -> {
+ // Clear back-off entries for every group whose plugin
state we
+ // attempted to delete (regardless of plugin outcome):
the group is
+ // about to be tombstoned on success and re-evaluated
by the next
+ // heartbeat on failure, so any in-flight back-off
entry at the old
+ // epoch is no longer load-bearing.
+
groupsWithStored.forEach(streamsGroupTopologyDescriptionManager::clearBackoffGroup);
+ return failures;
+ }))
+ .exceptionally(exception -> {
+ // Use ApiError.fromThrowable so the error reported to the
client matches what
+ // the sibling DeleteGroups paths return: it unwraps
CompletionException so the
+ // ErrorMessage is the cause's own message rather than the
FQCN-prefixed
+ // toString(), and it suppresses the message for
UNKNOWN_SERVER_ERROR so we do
+ // not leak plugin internals.
+ ApiError apiError = ApiError.fromThrowable(exception);
+ Map<String, ApiError> failures = new HashMap<>();
+ groupIds.forEach(id -> failures.put(id, apiError));
+ return failures;
+ });
+ }
+
+ /**
+ * Move plugin failures into {@code deletableGroupResults} and return the
group ids
+ * that should still proceed to tombstoning. Version-agnostic: the raw
{@link ApiError}
+ * is added as-is; any per-version translation of new error codes (e.g.
downgrading
+ * {@code GROUP_DELETION_FAILED} for {@code DeleteGroups} v<3) happens
at the
+ * {@code KafkaApis} layer where {@code request.context.apiVersion()} is
in scope and
+ * matches how other new error codes are version-gated.
+ */
+ private static List<String> filterStreamsTopologyErrors(
+ Map<String, ApiError> streamsErrMap,
+ List<String> groupIds,
+ DeleteGroupsResponseData.DeletableGroupResultCollection
deletableGroupResults
+ ) {
+ if (streamsErrMap.isEmpty()) {
+ return groupIds;
+ }
+ List<String> retained = new ArrayList<>();
+ for (String groupId : groupIds) {
+ ApiError err = streamsErrMap.get(groupId);
+ if (err == null) {
+ retained.add(groupId);
+ } else {
+ deletableGroupResults.add(
+ new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId(groupId)
+ .setErrorCode(err.error().code())
+ .setErrorMessage(err.message())
+ );
+ }
+ }
+ return retained;
+ }
+
private
CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection>
handleDeleteGroups(
AuthorizableRequestContext context,
TopicPartition topicPartition,
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
index ed25f77c158..d449f4b18f2 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
@@ -132,6 +132,7 @@ import org.apache.kafka.timeline.SnapshotRegistry;
import org.slf4j.Logger;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -958,6 +959,18 @@ public class GroupCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
return
groupMetadataManager.streamsGroupSetTopologyDescriptionEpoch(groupId,
pushedEpoch, permanentFailure);
}
+ /**
+ * Return the subset of {@code groupIds} that are streams groups with a
stored topology
+ * description. Used during {@code DeleteGroups} to identify which groups
need a
+ * {@code plugin.deleteTopology} call before being tombstoned.
+ */
+ public Set<String> streamsGroupsWithStoredTopologyDescription(
+ Collection<String> groupIds,
+ long committedOffset
+ ) {
+ return
groupMetadataManager.streamsGroupsWithStoredTopologyDescription(groupIds,
committedOffset);
+ }
+
/**
* Handles a ShareGroupDescribe request.
*
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
index 0c42f200529..5a32fa8f1cb 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
@@ -182,6 +182,7 @@ import org.slf4j.Logger;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -8390,6 +8391,35 @@ public class GroupMetadataManager {
return member;
}
+ /**
+ * Filter the given group ids down to those that are streams groups with a
non-default
+ * {@code StoredDescriptionTopologyEpoch}. The result is used by {@code
DeleteGroups}
+ * to decide which groups warrant a {@code plugin.deleteTopology} call
before the
+ * group is tombstoned. Non-existent groups and non-streams groups are
silently
+ * skipped — they have no plugin state to clean up.
+ *
+ * @param groupIds Candidate group ids on this shard.
+ * @param committedOffset A committed offset corresponding to the desired
snapshot.
+ * @return The subset of {@code groupIds} that are streams groups with a
stored topology.
+ */
+ public Set<String> streamsGroupsWithStoredTopologyDescription(
+ Collection<String> groupIds,
+ long committedOffset
+ ) {
+ Set<String> withStored = new HashSet<>();
+ for (String groupId : groupIds) {
+ try {
+ StreamsGroup group = streamsGroup(groupId, committedOffset);
+ if (group.storedDescriptionTopologyEpoch(committedOffset) !=
-1) {
+ withStored.add(groupId);
+ }
+ } catch (GroupIdNotFoundException ignored) {
+ // Not a streams group on this shard; nothing to clean up.
+ }
+ }
+ return withStored;
+ }
+
/**
* Persist the outcome of a topology description plugin call for a streams
group.
*
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java
index bf6f62dc804..9b52a1506d3 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java
@@ -18,14 +18,20 @@ package org.apache.kafka.coordinator.group.streams;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.ApiError;
import org.apache.kafka.common.requests.StreamsGroupHeartbeatResponse.Status;
import org.apache.kafka.common.utils.Time;
import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription;
import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescriptionPlugin;
import
org.apache.kafka.coordinator.group.api.streams.StreamsTopologyDescriptionPermanentFailureException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
/**
@@ -198,6 +204,61 @@ public class StreamsGroupTopologyDescriptionManager
implements AutoCloseable {
backoff.clearGroup(groupId);
}
+ /**
+ * Call {@code plugin.deleteTopology} for every supplied group id. Returns
a per-group
+ * map of failures keyed by group id; groups absent from the map either
had no plugin
+ * configured or the plugin call succeeded. The returned future never
completes
+ * exceptionally — failures are folded into the map so the service-level
+ * {@code DeleteGroups} flow can dispatch on the per-group outcome without
try/catch
+ * on the underlying future. A synchronous throw from the plugin (which
violates the
+ * SPI contract) is mapped to the same {@code GROUP_DELETION_FAILED} as an
+ * exceptional future.
+ *
+ * <p>Pure plugin invocation: does not read group state and does not touch
the
+ * back-off map. The service layer pre-filters the input via
+ * {@code streamsGroupsWithStoredTopologyDescription} and is responsible
for invoking
+ * {@link #clearBackoffGroup} for the groups that were attempted.
+ */
+ public CompletableFuture<Map<String, ApiError>>
invokeDeleteTopologies(Set<String> groupIds) {
+ if (plugin.isEmpty() || groupIds.isEmpty()) {
+ return CompletableFuture.completedFuture(Map.of());
+ }
+ final StreamsGroupTopologyDescriptionPlugin p = plugin.get();
+ List<CompletableFuture<Map.Entry<String, ApiError>>> outcomes = new
ArrayList<>(groupIds.size());
+ for (String groupId : groupIds) {
+ CompletableFuture<Map.Entry<String, ApiError>> outcome;
+ try {
+ outcome = p.deleteTopology(groupId).handle((unused, throwable)
-> toFailureEntry(groupId, throwable));
+ } catch (Exception e) {
+ // Synchronous throw from the plugin violates the SPI
contract; treat it as
+ // any other per-group failure so the failures map carries it
back to the
+ // caller without dropping the rest of the batch.
+ outcome =
CompletableFuture.completedFuture(toFailureEntry(groupId, e));
+ }
+ outcomes.add(outcome);
+ }
+ CompletableFuture<?>[] all = outcomes.toArray(new
CompletableFuture<?>[0]);
+ return CompletableFuture.allOf(all).thenApply(unused -> {
+ Map<String, ApiError> failures = new HashMap<>();
+ for (CompletableFuture<Map.Entry<String, ApiError>> future :
outcomes) {
+ Map.Entry<String, ApiError> entry = future.join();
+ if (entry != null) {
+ failures.put(entry.getKey(), entry.getValue());
+ }
+ }
+ return failures;
+ });
+ }
+
+ private static Map.Entry<String, ApiError> toFailureEntry(String groupId,
Throwable throwable) {
+ if (throwable == null) {
+ return null;
+ }
+ Throwable cause = Errors.maybeUnwrapException(throwable);
+ String message = cause != null ? cause.getMessage() : "Plugin failure
(no cause).";
+ return Map.entry(groupId, new ApiError(Errors.GROUP_DELETION_FAILED,
message));
+ }
+
// Visible for testing.
StreamsGroupTopologyDescriptionBackoff backoff() {
return backoff;
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java
index da4c4ae6591..20ac95683b6 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java
@@ -19,8 +19,10 @@ package org.apache.kafka.coordinator.group;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.GroupIdNotFoundException;
+import org.apache.kafka.common.errors.NotCoordinatorException;
import org.apache.kafka.common.errors.UnknownMemberIdException;
import org.apache.kafka.common.internals.Topic;
+import org.apache.kafka.common.message.DeleteGroupsResponseData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateRequestData;
@@ -28,6 +30,7 @@ import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateResp
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.internals.BufferSupplier;
import org.apache.kafka.common.utils.internals.LogContext;
import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
import org.apache.kafka.coordinator.common.runtime.CoordinatorRuntime;
@@ -44,6 +47,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.TimeUnit;
@@ -54,6 +58,7 @@ import static
org.apache.kafka.coordinator.group.GroupConfigManagerTest.createCo
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
@@ -360,6 +365,47 @@ public class
GroupCoordinatorServiceTopologyDescriptionTest {
assertTrue(heartbeatTopologyDescriptionRequired(runtime, service, 3,
-1, -1));
}
+ @Test
+ public void testUpdatePostPluginWriteRoutingFailureDoesNotArmBackoff()
throws Exception {
+ // The plugin succeeds, but between the plugin call and the
bookkeeping write this
+ // broker stops being the coordinator (NotCoordinatorException
surfaces from the
+ // write). The client will retry against the new coordinator, which
holds no
+ // back-off entry of its own; arming a broker-wide entry on this
broker would leak
+ // until expiry and could later suppress a legitimate solicitation if
the group
+ // migrates back. CoordinatorLoadInProgressException and
CoordinatorNotAvailableException
+ // travel the same NOOP branch — covered by one representative case to
avoid
+ // parameterized-test scaffolding.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(plugin.setTopology(anyString(), anyInt(), any()))
+ .thenReturn(CompletableFuture.completedFuture(null));
+ when(runtime.scheduleReadOperation(
+ eq("streams-group-topology-description-validate"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(null));
+ when(runtime.scheduleWriteOperation(
+ eq("streams-group-set-stored-topology-epoch"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.failedFuture(
+ new NotCoordinatorException("Lost coordinator status between
plugin success and bookkeeping write.")));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+
+ StreamsGroupTopologyDescriptionUpdateResponseData response =
service.streamsGroupTopologyDescriptionUpdate(
+ requestContext(ApiKeys.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_UPDATE),
+ validUpdateRequest()
+ ).get(5, TimeUnit.SECONDS);
+
+ assertEquals(Errors.NOT_COORDINATOR.code(), response.errorCode());
+ verify(plugin, times(1)).setTopology(eq("foo"), eq(3), any());
+
+ // Back-off must be untouched so the new coordinator (or this broker
after a
+ // migration back) can still solicit a fresh push at the same epoch.
+ assertTrue(heartbeatTopologyDescriptionRequired(runtime, service, 3,
-1, -1));
+ }
+
@Test
public void
testUpdateGroupDisappearsBetweenPluginSuccessAndWriteDropsBackoffEntry() throws
Exception {
// KIP-1331 race: the plugin succeeds, then the group is deleted, then
the bookkeeping
@@ -634,6 +680,215 @@ public class
GroupCoordinatorServiceTopologyDescriptionTest {
verify(plugin, times(1)).close();
}
+ @Test
+ public void testDeleteGroupsPluginFailureReturnsGroupDeletionFailed()
throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(plugin.deleteTopology("foo"))
+ .thenReturn(CompletableFuture.failedFuture(new
RuntimeException("plugin offline")));
+
+ when(runtime.scheduleWriteOperation(
+ eq("delete-share-groups"),
+ any(),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(Map.of()));
+ when(runtime.scheduleReadOperation(
+ eq("streams-group-topology-pre-delete"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(Set.of("foo")));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection results =
+ service.deleteGroups(
+ requestContext(ApiKeys.DELETE_GROUPS),
+ List.of("foo"),
+ BufferSupplier.NO_CACHING
+ ).get(5, TimeUnit.SECONDS);
+
+ DeleteGroupsResponseData.DeletableGroupResult result =
results.find("foo");
+ assertNotNull(result);
+ assertEquals(Errors.GROUP_DELETION_FAILED.code(), result.errorCode());
+ assertEquals("plugin offline", result.errorMessage());
+ verify(runtime, never()).scheduleWriteOperation(
+ eq("delete-groups"), any(), any());
+ }
+
+ @Test
+ public void testDeleteGroupsPluginSuccessProceedsToTombstone() throws
Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(plugin.deleteTopology("foo"))
+ .thenReturn(CompletableFuture.completedFuture(null));
+
+ when(runtime.scheduleWriteOperation(
+ eq("delete-share-groups"),
+ any(),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(Map.of()));
+ when(runtime.scheduleReadOperation(
+ eq("streams-group-topology-pre-delete"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(Set.of("foo")));
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection
tombstoneResult =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ tombstoneResult.add(new
DeleteGroupsResponseData.DeletableGroupResult().setGroupId("foo"));
+ when(runtime.scheduleWriteOperation(
+ eq("delete-groups"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(tombstoneResult));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection results =
+ service.deleteGroups(
+ requestContext(ApiKeys.DELETE_GROUPS),
+ List.of("foo"),
+ BufferSupplier.NO_CACHING
+ ).get(5, TimeUnit.SECONDS);
+
+ DeleteGroupsResponseData.DeletableGroupResult result =
results.find("foo");
+ assertNotNull(result);
+ assertEquals(Errors.NONE.code(), result.errorCode());
+ assertNull(result.errorMessage());
+ verify(plugin, times(1)).deleteTopology("foo");
+ verify(runtime, times(1)).scheduleWriteOperation(
+ eq("delete-groups"), eq(GROUP_TP), any());
+ }
+
+ @Test
+ public void testDeleteGroupsWithoutPluginSkipsPluginCall() throws
Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ when(runtime.scheduleWriteOperation(
+ eq("delete-share-groups"),
+ any(),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(Map.of()));
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection
tombstoneResult =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ tombstoneResult.add(new
DeleteGroupsResponseData.DeletableGroupResult().setGroupId("foo"));
+ when(runtime.scheduleWriteOperation(
+ eq("delete-groups"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(tombstoneResult));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.empty(), true);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection results =
+ service.deleteGroups(
+ requestContext(ApiKeys.DELETE_GROUPS),
+ List.of("foo"),
+ BufferSupplier.NO_CACHING
+ ).get(5, TimeUnit.SECONDS);
+
+ DeleteGroupsResponseData.DeletableGroupResult result =
results.find("foo");
+ assertNotNull(result);
+ assertEquals(Errors.NONE.code(), result.errorCode());
+ verify(runtime, never()).scheduleReadOperation(
+ eq("streams-group-topology-pre-delete"), any(), any());
+ verify(runtime, times(1)).scheduleWriteOperation(
+ eq("delete-groups"), eq(GROUP_TP), any());
+ }
+
+ @Test
+ public void testDeleteGroupsSkipsPluginCallWhenNoStoredTopology() throws
Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+
+ when(runtime.scheduleWriteOperation(
+ eq("delete-share-groups"),
+ any(),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(Map.of()));
+ when(runtime.scheduleReadOperation(
+ eq("streams-group-topology-pre-delete"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(Set.of()));
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection
tombstoneResult =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ tombstoneResult.add(new
DeleteGroupsResponseData.DeletableGroupResult().setGroupId("foo"));
+ when(runtime.scheduleWriteOperation(
+ eq("delete-groups"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(tombstoneResult));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection results =
+ service.deleteGroups(
+ requestContext(ApiKeys.DELETE_GROUPS),
+ List.of("foo"),
+ BufferSupplier.NO_CACHING
+ ).get(5, TimeUnit.SECONDS);
+
+ DeleteGroupsResponseData.DeletableGroupResult result =
results.find("foo");
+ assertNotNull(result);
+ assertEquals(Errors.NONE.code(), result.errorCode());
+ verify(plugin, never()).deleteTopology(anyString());
+ verify(runtime, times(1)).scheduleWriteOperation(
+ eq("delete-groups"), eq(GROUP_TP), any());
+ }
+
+ @Test
+ public void testDeleteGroupsMixedPluginOutcome() throws Exception {
+ // Two streams groups on the same partition; plugin succeeds for
"good", fails for "bad".
+ // Only "good" should reach the underlying delete-groups write; "bad"
surfaces as
+ // GROUP_DELETION_FAILED in the response.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(plugin.deleteTopology("good"))
+ .thenReturn(CompletableFuture.completedFuture(null));
+ when(plugin.deleteTopology("bad"))
+ .thenReturn(CompletableFuture.failedFuture(new
RuntimeException("rejected")));
+
+ when(runtime.scheduleWriteOperation(
+ eq("delete-share-groups"),
+ any(),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(Map.of()));
+ when(runtime.scheduleReadOperation(
+ eq("streams-group-topology-pre-delete"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(Set.of("good",
"bad")));
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection
tombstoneResult =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ tombstoneResult.add(new
DeleteGroupsResponseData.DeletableGroupResult().setGroupId("good"));
+ when(runtime.scheduleWriteOperation(
+ eq("delete-groups"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(tombstoneResult));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection results =
+ service.deleteGroups(
+ requestContext(ApiKeys.DELETE_GROUPS),
+ List.of("good", "bad"),
+ BufferSupplier.NO_CACHING
+ ).get(5, TimeUnit.SECONDS);
+
+ DeleteGroupsResponseData.DeletableGroupResult goodResult =
results.find("good");
+ assertNotNull(goodResult);
+ assertEquals(Errors.NONE.code(), goodResult.errorCode());
+
+ DeleteGroupsResponseData.DeletableGroupResult badResult =
results.find("bad");
+ assertNotNull(badResult);
+ assertEquals(Errors.GROUP_DELETION_FAILED.code(),
badResult.errorCode());
+ assertEquals("rejected", badResult.errorMessage());
+ }
+
private static StreamsGroupTopologyDescriptionUpdateRequestData
validUpdateRequest() {
return new StreamsGroupTopologyDescriptionUpdateRequestData()
.setGroupId("foo")
