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 111a4685df8 KAFKA-20623: Update RPC for streams group topology
description plugin (2/3) (#22552)
111a4685df8 is described below
commit 111a4685df81f5566f797d1a294bbc9ea0a9af01
Author: TengYao Chi <[email protected]>
AuthorDate: Tue Jun 16 19:29:03 2026 +0100
KAFKA-20623: Update RPC for streams group topology description plugin (2/3)
(#22552)
JIRA: KAFKA-20623
This is a part of KIP-1331
**This PR shouldn't be merge before #22551**
Adds the StreamsGroupTopologyDescriptionUpdate RPC handler stacked on
the heartbeat extension (1/3). The push pipeline runs through
`TopologyDescriptionManager.handleSetTopology`: `validate (group,
member)`, convert wire payload, call `plugin.setTopology`, persist
`StoredDescriptionTopologyEpoch` on success or
`FailedDescriptionTopologyEpoch` on permanent failure, and centralise
back-off state mutations in a single `whenComplete`.
### Push pipeline on the manager
`TopologyDescriptionManager` gains the push entry points:
- `preCheckTopologyDescriptionUpdate` — synchronous structural
validation that rejects
the request with `UNSUPPORTED_VERSION` when no plugin is configured,
`INVALID_REQUEST`
for empty `MemberId` / `GroupId` or null `TopologyDescription`.
- `handleSetTopology` — runs `validateStreamsGroupMember` first (so a
fenced caller gets
`UNKNOWN_MEMBER_ID` rather than an `INVALID_REQUEST` from a payload
conversion
failure), then converts the wire payload to the broker-side POJO, then
calls the
plugin. Plugin success writes a metadata record advancing
`StoredDescriptionTopologyEpoch`; a
`StreamsTopologyDescriptionPermanentFailureException`
(or a synchronous throw) writes `FailedDescriptionTopologyEpoch`; any
other exception
is treated as transient and writes no record.
### Back-off mutation point
All back-off mutations on the push path are folded into a single
`whenComplete`,
driven by a `BackoffAction` holder populated by each terminal branch.
The default is
`ARM`; only success and permanent-failure branches set `CLEAR`. A
post-plugin write
failure therefore re-arms the back-off and the next heartbeat sees the
drift and
re-solicits an idempotent re-push, matching the KIP-1331 invariant.
### Wire ↔ POJO converter
`StreamsGroupTopologyDescriptionConverter` translates
`StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription`
into the
broker-side `StreamsGroupTopologyDescription` POJO. Subtopology node
ordering and
string-collection iteration order are preserved via `LinkedHashSet`, so
a downstream
pretty-printer (e.g. the future `kafka-streams-groups.sh --topology`
command) can
reproduce the source ordering.
Wire-level node types (`SOURCE=1`, `PROCESSOR=2`, `SINK=3`) map to the
sealed `Node` hierarchy. `GlobalStore` shape is validated; a malformed
payload throws `InvalidRequestException`. The POJO itself drops
`Set.copyOf` in favour of `Collections.unmodifiableSet` to preserve the
converter's insertion order — production callers only ever construct
nodes from a fresh `LinkedHashSet` so this is not a defensive-copy
regression.
### Coordinator interface and service entry
- `GroupCoordinator.streamsGroupTopologyDescriptionUpdate(context,
request)` interface
method added.
- `GroupCoordinatorService.streamsGroupTopologyDescriptionUpdate(...)`
short-circuits
on a non-active coordinator with `COORDINATOR_NOT_AVAILABLE`, then
delegates to
`topologyDescriptionManager.preCheckTopologyDescriptionUpdate(...)`
and
`handleSetTopology(...)`. Unhandled exceptions are translated by
`handleOperationException` into the wire error response.
### Coordinator shard + GMM
- `GroupMetadataManager.streamsGroupSetTopologyDescriptionEpoch(groupId,
pushedEpoch, permanentFailure)`
— writes a `StreamsGroupMetadata` record advancing
`StoredDescriptionTopologyEpoch`
(on success) or `FailedDescriptionTopologyEpoch` (on permanent
failure).
- `GroupCoordinatorShard.streamsGroupSetTopologyDescriptionEpoch(...)`
exposes the
method to the runtime write-operation scheduler.
`topologyDescriptionManager.preCheckTopologyDescriptionUpdate(...)`
and
`handleSetTopology(...)`. Unhandled exceptions are translated by
`handleOperationException` into the wire error response.
### Coordinator shard + GMM
- `GroupMetadataManager.streamsGroupSetTopologyDescriptionEpoch(groupId,
pushedEpoch, permanentFailure)`
— writes a `StreamsGroupMetadata` record advancing
`StoredDescriptionTopologyEpoch`
(on success) or `FailedDescriptionTopologyEpoch` (on permanent
failure).
- `GroupCoordinatorShard.streamsGroupSetTopologyDescriptionEpoch(...)`
exposes the
method to the runtime write-operation scheduler.
Reviewers: Lucas Brutschy <[email protected]>
---
.../streams/StreamsGroupTopologyDescription.java | 11 +-
.../kafka/coordinator/group/GroupCoordinator.java | 23 ++
.../coordinator/group/GroupCoordinatorService.java | 191 +++++++++-
.../coordinator/group/GroupCoordinatorShard.java | 36 +-
.../coordinator/group/GroupMetadataManager.java | 100 +++--
.../group/streams/StreamsGroupHeartbeatResult.java | 24 +-
.../StreamsGroupTopologyDescriptionBackoff.java | 108 ++++--
.../StreamsGroupTopologyDescriptionConverter.java | 108 ++++++
.../StreamsGroupTopologyDescriptionManager.java | 115 +++++-
...pCoordinatorServiceTopologyDescriptionTest.java | 411 ++++++++++++++++++++-
.../group/GroupCoordinatorShardTest.java | 19 +-
.../group/GroupMetadataManagerTest.java | 58 ++-
.../StreamsGroupTopologyDescriptionTest.java | 9 +-
...StreamsGroupTopologyDescriptionBackoffTest.java | 146 ++++++--
...reamsGroupTopologyDescriptionConverterTest.java | 197 ++++++++++
15 files changed, 1398 insertions(+), 158 deletions(-)
diff --git
a/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/StreamsGroupTopologyDescription.java
b/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/StreamsGroupTopologyDescription.java
index bfff3aca4f6..a5f242ed5a5 100644
---
a/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/StreamsGroupTopologyDescription.java
+++
b/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/StreamsGroupTopologyDescription.java
@@ -19,6 +19,7 @@ package org.apache.kafka.coordinator.group.api.streams;
import org.apache.kafka.common.annotation.InterfaceStability;
import java.util.Collection;
+import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
@@ -58,16 +59,16 @@ public record StreamsGroupTopologyDescription(
public record Source(String name, Set<String> topics, Set<String>
successors) implements Node {
public Source {
Objects.requireNonNull(name, "name");
- topics = Set.copyOf(Objects.requireNonNull(topics, "topics"));
- successors = Set.copyOf(Objects.requireNonNull(successors,
"successors"));
+ topics =
Collections.unmodifiableSet(Objects.requireNonNull(topics, "topics"));
+ successors =
Collections.unmodifiableSet(Objects.requireNonNull(successors, "successors"));
}
}
public record Processor(String name, Set<String> stores, Set<String>
successors) implements Node {
public Processor {
Objects.requireNonNull(name, "name");
- stores = Set.copyOf(Objects.requireNonNull(stores, "stores"));
- successors = Set.copyOf(Objects.requireNonNull(successors,
"successors"));
+ stores =
Collections.unmodifiableSet(Objects.requireNonNull(stores, "stores"));
+ successors =
Collections.unmodifiableSet(Objects.requireNonNull(successors, "successors"));
}
}
@@ -75,7 +76,7 @@ public record StreamsGroupTopologyDescription(
public Sink {
Objects.requireNonNull(name, "name");
Objects.requireNonNull(topic, "topic");
- successors = Set.copyOf(Objects.requireNonNull(successors,
"successors"));
+ successors =
Collections.unmodifiableSet(Objects.requireNonNull(successors, "successors"));
}
}
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java
index c6f67ade887..717664a5939 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java
@@ -47,6 +47,8 @@ import
org.apache.kafka.common.message.ShareGroupHeartbeatRequestData;
import org.apache.kafka.common.message.ShareGroupHeartbeatResponseData;
import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
+import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateRequestData;
+import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateResponseData;
import org.apache.kafka.common.message.SyncGroupRequestData;
import org.apache.kafka.common.message.SyncGroupResponseData;
import org.apache.kafka.common.message.TxnOffsetCommitRequestData;
@@ -99,6 +101,27 @@ public interface GroupCoordinator {
StreamsGroupHeartbeatRequestData request
);
+ /**
+ * Persist the full topology description pushed by a Streams group member.
+ *
+ * <p>The broker forwards the description to the configured
+ * {@code StreamsGroupTopologyDescriptionPlugin}. On success the broker
writes a
+ * metadata record advancing {@code StoredDescriptionTopologyEpoch}. On a
permanent
+ * plugin failure it writes {@code FailedDescriptionTopologyEpoch} to stop
re-soliciting
+ * at the same epoch. On a transient failure no record is written and the
broker arms
+ * an in-memory back-off.
+ *
+ * @param context The request context.
+ * @param request The StreamsGroupTopologyDescriptionUpdateRequest data.
+ *
+ * @return A future yielding the response. The error code is set to
indicate any error
+ * encountered during the execution.
+ */
+ CompletableFuture<StreamsGroupTopologyDescriptionUpdateResponseData>
streamsGroupTopologyDescriptionUpdate(
+ AuthorizableRequestContext context,
+ StreamsGroupTopologyDescriptionUpdateRequestData request
+ );
+
/**
* Heartbeat to a Share Group.
*
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 303a7d1f2e0..1f78c0b401f 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,10 +20,12 @@ 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.GroupIdNotFoundException;
import org.apache.kafka.common.errors.InvalidRequestException;
import org.apache.kafka.common.errors.NotCoordinatorException;
import org.apache.kafka.common.errors.StreamsInvalidTopologyException;
import org.apache.kafka.common.errors.UnsupportedAssignorException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.internals.Plugin;
import org.apache.kafka.common.internals.Topic;
import org.apache.kafka.common.message.AlterShareGroupOffsetsRequestData;
@@ -59,6 +61,8 @@ import
org.apache.kafka.common.message.ShareGroupHeartbeatResponseData;
import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
+import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateRequestData;
+import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateResponseData;
import org.apache.kafka.common.message.SyncGroupRequestData;
import org.apache.kafka.common.message.SyncGroupResponseData;
import org.apache.kafka.common.message.TxnOffsetCommitRequestData;
@@ -102,6 +106,7 @@ import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescri
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics;
import org.apache.kafka.coordinator.group.streams.StreamsGroupDescribeResult;
import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
+import
org.apache.kafka.coordinator.group.streams.StreamsGroupTopologyDescriptionConverter;
import
org.apache.kafka.coordinator.group.streams.StreamsGroupTopologyDescriptionManager;
import org.apache.kafka.image.MetadataDelta;
import org.apache.kafka.image.MetadataImage;
@@ -626,7 +631,7 @@ public class GroupCoordinatorService implements
GroupCoordinator {
) {
if (!isActive.get()) {
return CompletableFuture.completedFuture(
- new StreamsGroupHeartbeatResult(
+ StreamsGroupHeartbeatResult.withoutEpochContext(
new
StreamsGroupHeartbeatResponseData().setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())
)
);
@@ -638,7 +643,7 @@ public class GroupCoordinatorService implements
GroupCoordinator {
} catch (Throwable ex) {
ApiError apiError = ApiError.fromThrowable(ex);
return CompletableFuture.completedFuture(
- new StreamsGroupHeartbeatResult(
+ StreamsGroupHeartbeatResult.withoutEpochContext(
new StreamsGroupHeartbeatResponseData()
.setErrorCode(apiError.error().code())
.setErrorMessage(apiError.message())
@@ -646,17 +651,36 @@ public class GroupCoordinatorService implements
GroupCoordinator {
);
}
- return runtime.scheduleWriteOperation(
+ CompletableFuture<StreamsGroupHeartbeatResult> heartbeat =
runtime.scheduleWriteOperation(
"streams-group-heartbeat",
topicPartitionFor(request.groupId()),
- coordinator -> coordinator.streamsGroupHeartbeat(context, request)
- ).thenApply(result ->
streamsGroupTopologyDescriptionManager.maybeSetTopologyDescriptionRequired(result,
request.groupId(), context.requestVersion())
- ).exceptionally(exception -> handleOperationException(
+ coordinator -> coordinator.streamsGroupHeartbeat(context,
request));
+
+ if (streamsGroupTopologyDescriptionManager.isPluginConfigured()) {
+ heartbeat = heartbeat.thenApply(result -> {
+ try {
+ return
streamsGroupTopologyDescriptionManager.maybeSetTopologyDescriptionRequired(
+ result, request.groupId(), context.requestVersion());
+ } catch (Throwable t) {
+ // The heartbeat has already committed durably; if
decoration fails (e.g.
+ // because of an unexpected response shape) we log and
return the
+ // committed result as-is rather than translating into an
error via the
+ // exceptionally below — that would mask a successful
broker-side state
+ // change behind a client-visible failure.
+ log.warn("Failed to apply topology-description
post-processing on the "
+ + "streams group heartbeat response for group {};
returning the response unmodified.",
+ request.groupId(), t);
+ return result;
+ }
+ });
+ }
+
+ return heartbeat.exceptionally(exception -> handleOperationException(
"streams-group-heartbeat",
request,
exception,
(error, message) ->
- new StreamsGroupHeartbeatResult(
+ StreamsGroupHeartbeatResult.withoutEpochContext(
new StreamsGroupHeartbeatResponseData()
.setErrorCode(error.code())
.setErrorMessage(message)
@@ -665,6 +689,159 @@ public class GroupCoordinatorService implements
GroupCoordinator {
));
}
+ /**
+ * See {@link
GroupCoordinator#streamsGroupTopologyDescriptionUpdate(AuthorizableRequestContext,
StreamsGroupTopologyDescriptionUpdateRequestData)}.
+ *
+ * <p>The push pipeline lives on {@link TopologyDescriptionManager}; the
service is
+ * responsible only for short-circuiting on a non-active coordinator and
translating
+ * unhandled exceptions into the wire error response.
+ */
+ @Override
+ public
CompletableFuture<StreamsGroupTopologyDescriptionUpdateResponseData>
streamsGroupTopologyDescriptionUpdate(
+ AuthorizableRequestContext context,
+ StreamsGroupTopologyDescriptionUpdateRequestData request
+ ) {
+ if (!isActive.get()) {
+ return CompletableFuture.completedFuture(
+ new StreamsGroupTopologyDescriptionUpdateResponseData()
+ .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())
+ );
+ }
+
+ try {
+ throwIfStreamsGroupTopologyDescriptionUpdateInvalid(request);
+ } catch (Throwable ex) {
+ ApiError apiError = ApiError.fromThrowable(ex);
+ return CompletableFuture.completedFuture(new
StreamsGroupTopologyDescriptionUpdateResponseData()
+ .setErrorCode(apiError.error().code())
+ .setErrorMessage(apiError.message())
+ );
+ }
+
+ final String groupId = request.groupId();
+ final String memberId = request.memberId();
+ final int pushedEpoch = request.topologyEpoch();
+ final TopicPartition tp = topicPartitionFor(groupId);
+
+ // Each terminal branch produces a SetTopologyOutcome carrying the
response and the
+ // back-off disposition. Pre-plugin failures (validate / convert /
runtime error)
+ // skip the post-plugin stages and are wrapped with BackoffAction.NOOP
by
+ // exceptionally so a fenced or unauthorized caller cannot grief the
back-off and
+ // suppress legitimate solicitation. Post-plugin failures arm the
back-off; a
+ // post-plugin write failing with GroupIdNotFoundException — the group
was deleted
+ // between the plugin call and the write — drops the orphaned entry
since no live
+ // group remains to throttle.
+ return runtime.scheduleReadOperation(
+ "streams-group-topology-description-validate",
+ tp,
+ (coordinator, lastCommittedOffset) -> {
+ coordinator.validateStreamsGroupTopologyDescriptionUpdate(
+ groupId, memberId, pushedEpoch, lastCommittedOffset);
+ return null;
+ })
+ .thenApply(__ ->
StreamsGroupTopologyDescriptionConverter.fromRequest(request.topologyDescription()))
+ .thenCompose(description ->
streamsGroupTopologyDescriptionManager.invokeSetTopology(
+ groupId, pushedEpoch, description))
+ .thenCompose(pluginOutcome ->
postPluginSetTopologyAction(pluginOutcome, groupId, pushedEpoch, tp))
+ .exceptionally(t -> new SetTopologyOutcome(null,
BackoffAction.NOOP, t))
+ .thenApply(outcome -> {
+ applySetTopologyBackoff(outcome, groupId, pushedEpoch);
+ return outcome;
+ })
+ .thenCompose(outcome -> outcome.failure() == null
+ ? CompletableFuture.completedFuture(outcome.response())
+ : CompletableFuture.failedFuture(outcome.failure()))
+ .exceptionally(exception -> handleOperationException(
+ "streams-group-topology-description-update",
+ request,
+ exception,
+ (error, message) -> new
StreamsGroupTopologyDescriptionUpdateResponseData()
+ .setErrorCode(error.code())
+ .setErrorMessage(message),
+ log
+ ));
+ }
+
+ private CompletableFuture<SetTopologyOutcome> postPluginSetTopologyAction(
+ StreamsGroupTopologyDescriptionManager.PluginOutcome pluginOutcome,
+ String groupId,
+ int pushedEpoch,
+ TopicPartition tp
+ ) {
+ return switch (pluginOutcome.kind()) {
+ case SUCCESS -> runtime.scheduleWriteOperation(
+ "streams-group-set-stored-topology-epoch",
+ tp,
+ coordinator ->
coordinator.streamsGroupSetTopologyDescriptionEpoch(groupId, pushedEpoch, false)
+ ).handle((unused, throwable) -> outcomeForPostPluginWrite(
+ throwable, new
StreamsGroupTopologyDescriptionUpdateResponseData()));
+ case PERMANENT -> runtime.scheduleWriteOperation(
+ "streams-group-set-failed-topology-epoch",
+ tp,
+ coordinator ->
coordinator.streamsGroupSetTopologyDescriptionEpoch(groupId, pushedEpoch, true)
+ ).handle((unused, throwable) -> outcomeForPostPluginWrite(
+ throwable,
+
topologyDescriptionUpdateError(Errors.STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED,
pluginOutcome.message())));
+ case TRANSIENT -> CompletableFuture.completedFuture(new
SetTopologyOutcome(
+
topologyDescriptionUpdateError(Errors.STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED,
pluginOutcome.message()),
+ BackoffAction.ARM, null));
+ };
+ }
+
+ private static SetTopologyOutcome outcomeForPostPluginWrite(
+ Throwable throwable,
+ StreamsGroupTopologyDescriptionUpdateResponseData responseOnSuccess
+ ) {
+ if (throwable == null) {
+ return new SetTopologyOutcome(responseOnSuccess,
BackoffAction.CLEAR, null);
+ }
+ // 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) {
+ return new SetTopologyOutcome(null, BackoffAction.CLEAR_GROUP,
throwable);
+ }
+ return new SetTopologyOutcome(null, BackoffAction.ARM, throwable);
+ }
+
+ private void applySetTopologyBackoff(SetTopologyOutcome outcome, String
groupId, int pushedEpoch) {
+ switch (outcome.backoffAction()) {
+ case NOOP -> { }
+ case CLEAR ->
streamsGroupTopologyDescriptionManager.clearBackoff(groupId, pushedEpoch);
+ case CLEAR_GROUP ->
streamsGroupTopologyDescriptionManager.clearBackoffGroup(groupId);
+ case ARM ->
streamsGroupTopologyDescriptionManager.armBackoff(groupId, pushedEpoch);
+ }
+ }
+
+ private static StreamsGroupTopologyDescriptionUpdateResponseData
topologyDescriptionUpdateError(
+ Errors error,
+ String message
+ ) {
+ return new StreamsGroupTopologyDescriptionUpdateResponseData()
+ .setErrorCode(error.code())
+ .setErrorMessage(message);
+ }
+
+ private record SetTopologyOutcome(
+ StreamsGroupTopologyDescriptionUpdateResponseData response,
+ BackoffAction backoffAction,
+ Throwable failure
+ ) { }
+
+ private enum BackoffAction { NOOP, ARM, CLEAR, CLEAR_GROUP }
+
+ private void throwIfStreamsGroupTopologyDescriptionUpdateInvalid(
+ StreamsGroupTopologyDescriptionUpdateRequestData request
+ ) throws InvalidRequestException, UnsupportedVersionException {
+ if (!streamsGroupTopologyDescriptionManager.isPluginConfigured()) {
+ throw new UnsupportedVersionException(
+ "The broker has no streams group topology description plugin
configured.");
+ }
+ throwIfEmptyString(request.memberId(), "MemberId can't be empty.");
+ throwIfEmptyString(request.groupId(), "GroupId can't be empty.");
+ throwIfNull(request.topologyDescription(), "TopologyDescription can't
be null.");
+ }
+
/**
* Validates the ShareGroupHeartbeat request.
*
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 b65260f1123..ed25f77c158 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
@@ -21,6 +21,7 @@ import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.ApiException;
import org.apache.kafka.common.errors.GroupIdNotFoundException;
import org.apache.kafka.common.errors.GroupNotEmptyException;
+import org.apache.kafka.common.errors.InvalidRequestException;
import org.apache.kafka.common.errors.UnknownMemberIdException;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.internals.Plugin;
@@ -913,23 +914,48 @@ public class GroupCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
}
/**
- * Validates that a streams group exists and that the given member is a
current member of it.
- * The lookup runs at {@code committedOffset} so an uncommitted
fence/leave does not
- * cause a still-live member to appear unknown (or vice versa). Must be
scheduled on the
+ * Validates that a streams group exists, that the given member is a
current member of
+ * it, and that {@code pushedEpoch} matches the group's current topology
epoch. The
+ * lookup runs at {@code committedOffset} so an uncommitted fence/leave
does not cause
+ * a still-live member to appear unknown (or vice versa). Must be
scheduled on the
* coordinator runtime like any other read.
*
* @param groupId The group ID.
* @param memberId The member ID.
+ * @param pushedEpoch The topology epoch carried by the push request.
* @param committedOffset A committed offset corresponding to the desired
snapshot.
* @throws GroupIdNotFoundException if the group does not exist.
* @throws UnknownMemberIdException if the member is not in the group.
+ * @throws InvalidRequestException if {@code pushedEpoch} does not match
the group's
+ * current topology epoch.
*/
- public void validateStreamsGroupMember(
+ public void validateStreamsGroupTopologyDescriptionUpdate(
String groupId,
String memberId,
+ int pushedEpoch,
long committedOffset
) {
- groupMetadataManager.validateStreamsGroupMember(groupId, memberId,
committedOffset);
+ groupMetadataManager.validateStreamsGroupTopologyDescriptionUpdate(
+ groupId, memberId, pushedEpoch, committedOffset);
+ }
+
+ /**
+ * Persist the outcome of a topology description plugin call. Writes a
metadata record
+ * advancing either {@code StoredDescriptionTopologyEpoch} (on plugin
success) or
+ * {@code FailedDescriptionTopologyEpoch} (on permanent plugin failure).
+ *
+ * @param groupId The streams group id.
+ * @param pushedEpoch The topology epoch on the push that just
completed.
+ * @param permanentFailure True if the plugin signalled a permanent
failure; false on success.
+ * @return A coordinator result carrying the metadata record.
+ * @throws GroupIdNotFoundException if the streams group no longer exists.
+ */
+ public CoordinatorResult<Void, CoordinatorRecord>
streamsGroupSetTopologyDescriptionEpoch(
+ String groupId,
+ int pushedEpoch,
+ boolean permanentFailure
+ ) {
+ return
groupMetadataManager.streamsGroupSetTopologyDescriptionEpoch(groupId,
pushedEpoch, permanentFailure);
}
/**
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 c5673613fe4..0c42f200529 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
@@ -4469,16 +4469,15 @@ public class GroupMetadataManager {
.setMemberEpoch(memberEpoch)
.setStatus(List.of());
+ // Leave/fence paths use
StreamsGroupHeartbeatResult.withoutEpochContext: the departing
+ // member will never push a topology description, so the service-layer
post-processing
+ // sees epochs of -1 and short-circuits. Without this the manager
would arm a back-off
+ // window for a member that is on its way out, delaying push
solicitation for the rest
+ // of the group.
if (instanceId == null) {
StreamsGroupMember member = group.getMemberOrThrow(memberId);
log.info("[GroupId {}][MemberId {}] Member {} left the streams
group.", groupId, memberId, memberId);
- return streamsGroupFenceMember(group, member, new
StreamsGroupHeartbeatResult(
- response,
- Map.of(),
- group.currentTopologyEpoch(),
- group.storedDescriptionTopologyEpoch(),
- group.failedDescriptionTopologyEpoch()
- ));
+ return streamsGroupFenceMember(group, member,
StreamsGroupHeartbeatResult.withoutEpochContext(response));
} else {
StreamsGroupMember member = group.staticMember(instanceId);
throwIfStaticMemberIsUnknown(member, instanceId);
@@ -4490,13 +4489,7 @@ public class GroupMetadataManager {
} else {
log.info("[GroupId {}][MemberId {}] Static member {} with
instance id {} left the streams group.",
group.groupId(), memberId, memberId, instanceId);
- return streamsGroupFenceMember(group, member, new
StreamsGroupHeartbeatResult(
- response,
- Map.of(),
- group.currentTopologyEpoch(),
- group.storedDescriptionTopologyEpoch(),
- group.failedDescriptionTopologyEpoch()
- ));
+ return streamsGroupFenceMember(group, member,
StreamsGroupHeartbeatResult.withoutEpochContext(response));
}
}
}
@@ -4563,15 +4556,12 @@ public class GroupMetadataManager {
.setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
.setStatus(List.of());
+ // Static-leave is a departing path: a member that will not push at
this epoch,
+ // so we use the withoutEpochContext factory to skip the heartbeat
post-processing
+ // and avoid arming the back-off on its behalf.
return new CoordinatorResult<>(
List.of(record),
- new StreamsGroupHeartbeatResult(
- response,
- Map.of(),
- group.currentTopologyEpoch(),
- group.storedDescriptionTopologyEpoch(),
- group.failedDescriptionTopologyEpoch()
- )
+ StreamsGroupHeartbeatResult.withoutEpochContext(response)
);
}
@@ -8362,26 +8352,78 @@ public class GroupMetadataManager {
}
/**
- * Validates that a streams group exists and that the given member is a
current member of it.
- * Used by the StreamsGroupTopologyDescriptionUpdate RPC handler to
enforce the
- * GROUP_ID_NOT_FOUND / UNKNOWN_MEMBER_ID contract before consulting the
topology description plugin.
- * The lookup runs at {@code committedOffset} so an uncommitted
fence/leave does not cause a
- * still-live member to appear unknown (or vice versa).
+ * Validates that a streams group exists, that the given member is a
current member of it,
+ * and that {@code pushedEpoch} matches the group's current topology
epoch. Used by the
+ * StreamsGroupTopologyDescriptionUpdate RPC handler to enforce the
+ * GROUP_ID_NOT_FOUND / UNKNOWN_MEMBER_ID / INVALID_REQUEST contract
before consulting
+ * the topology description plugin.
+ *
+ * <p>The epoch check rejects stale or out-of-order pushes so they cannot
regress
+ * {@code storedDescriptionTopologyEpoch} or be persisted at an epoch the
group never
+ * advertised. The lookup runs at {@code committedOffset} so an uncommitted
+ * fence/leave does not cause a still-live member to appear unknown (or
vice versa).
*
* @param groupId The group ID.
* @param memberId The member ID.
+ * @param pushedEpoch The topology epoch carried by the push request.
* @param committedOffset A committed offset corresponding to the desired
snapshot.
* @return The matching {@link StreamsGroupMember}.
* @throws GroupIdNotFoundException if no streams group with this id
exists.
* @throws UnknownMemberIdException if the member is not currently in the
group.
+ * @throws InvalidRequestException if {@code pushedEpoch} does not match
the group's
+ * current topology epoch.
*/
- public StreamsGroupMember validateStreamsGroupMember(
+ public StreamsGroupMember validateStreamsGroupTopologyDescriptionUpdate(
String groupId,
String memberId,
+ int pushedEpoch,
long committedOffset
- ) throws GroupIdNotFoundException, UnknownMemberIdException {
+ ) throws GroupIdNotFoundException, UnknownMemberIdException,
InvalidRequestException {
StreamsGroup group = streamsGroup(groupId, committedOffset);
- return group.getMemberOrThrow(memberId, committedOffset);
+ StreamsGroupMember member = group.getMemberOrThrow(memberId,
committedOffset);
+ int currentEpoch = group.currentTopologyEpoch();
+ if (pushedEpoch != currentEpoch) {
+ throw new InvalidRequestException(
+ "Topology epoch " + pushedEpoch + " does not match the group's
current topology epoch "
+ + currentEpoch + ".");
+ }
+ return member;
+ }
+
+ /**
+ * Persist the outcome of a topology description plugin call for a streams
group.
+ *
+ * <p>On a successful plugin {@code setTopology} the {@code
StoredDescriptionTopologyEpoch}
+ * field is advanced to the pushed epoch; on a permanent failure the
+ * {@code FailedDescriptionTopologyEpoch} field is advanced instead so
subsequent
+ * heartbeats at the same epoch do not re-solicit a push.
+ *
+ * @param groupId The streams group id.
+ * @param pushedEpoch The topology epoch on the push that just
completed.
+ * @param permanentFailure True if the plugin signalled a permanent
failure; false on success.
+ * @return A coordinator result carrying the metadata record that updates
the field.
+ * @throws GroupIdNotFoundException if the streams group no longer exists.
+ */
+ public CoordinatorResult<Void, CoordinatorRecord>
streamsGroupSetTopologyDescriptionEpoch(
+ String groupId,
+ int pushedEpoch,
+ boolean permanentFailure
+ ) throws GroupIdNotFoundException {
+ StreamsGroup group = streamsGroup(groupId);
+
+ int newStored = permanentFailure ?
group.storedDescriptionTopologyEpoch() : pushedEpoch;
+ int newFailed = permanentFailure ? pushedEpoch :
group.failedDescriptionTopologyEpoch();
+
+ CoordinatorRecord record = newStreamsGroupMetadataRecord(
+ groupId,
+ group.groupEpoch(),
+ group.metadataHash(),
+ group.validatedTopologyEpoch(),
+ group.lastAssignmentConfigs(),
+ newStored,
+ newFailed
+ );
+ return new CoordinatorResult<>(List.of(record), null);
}
/**
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupHeartbeatResult.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupHeartbeatResult.java
index 82b84d7b494..28fc1e844f2 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupHeartbeatResult.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupHeartbeatResult.java
@@ -55,11 +55,25 @@ public record StreamsGroupHeartbeatResult(
}
/**
- * Convenience constructor for failure-fast paths that do not resolve a
group: no
- * internal topics to create, and all three epoch fields set to -1 so the
service-layer
- * gate sees nothing to do.
+ * Build a heartbeat result that bypasses the service-layer topology
post-processing.
+ * Used at two distinct call sites:
+ *
+ * <ul>
+ * <li>Failure-fast paths in {@code
GroupCoordinatorService#streamsGroupHeartbeat}
+ * — broker not active, request validation rejected, runtime error
translated
+ * by {@code handleOperationException}. The group is not resolved,
so there is
+ * no epoch context to track.</li>
+ * <li>Departing-member paths in {@code GroupMetadataManager} (leave or
fence).
+ * The group exists but the member will not push a topology
description, so
+ * attaching the live epoch would arm a back-off window on its
behalf and
+ * delay solicitation for the rest of the group.</li>
+ * </ul>
+ *
+ * <p>All three epoch fields are set to -1, causing
+ * {@code maybeSetTopologyDescriptionRequired} to short-circuit before
arming the
+ * back-off or setting the {@code TopologyDescriptionRequired} flag.
*/
- public StreamsGroupHeartbeatResult(StreamsGroupHeartbeatResponseData data)
{
- this(data, Map.of(), -1, -1, -1);
+ public static StreamsGroupHeartbeatResult
withoutEpochContext(StreamsGroupHeartbeatResponseData data) {
+ return new StreamsGroupHeartbeatResult(data, Map.of(), -1, -1, -1);
}
}
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionBackoff.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionBackoff.java
index afda7817908..2d4035aea95 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionBackoff.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionBackoff.java
@@ -17,6 +17,7 @@
package org.apache.kafka.coordinator.group.streams;
import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.internals.ExponentialBackoff;
import java.util.concurrent.ConcurrentHashMap;
@@ -24,22 +25,31 @@ import java.util.concurrent.ConcurrentHashMap;
* In-memory per-group back-off that throttles broker re-solicitation of a
topology
* description push. An entry is armed when the broker decides to set
* {@code TopologyDescriptionRequired=true} on a heartbeat or after a
transient plugin
- * failure; consecutive arms at the same topology epoch double the window from
- * {@value #INITIAL_DELAY_MS} ms up to {@value #MAX_DELAY_MS} ms. Successful
pushes,
- * permanent plugin failures, and topology-epoch advances clear the entry.
+ * failure; consecutive arms at the same topology epoch advance the attempt
count and
+ * delegate to {@link ExponentialBackoff} for the next delay . Successful
pushes, permanent plugin failures,
+ * and topology-epoch advances clear the entry.
*/
public class StreamsGroupTopologyDescriptionBackoff {
static final long INITIAL_DELAY_MS = 30_000L;
static final long MAX_DELAY_MS = 3_600_000L;
+ static final int MULTIPLIER = 2;
+ static final double JITTER = 0.2;
private final Time time;
+ private final ExponentialBackoff exponentialBackoff;
private final ConcurrentHashMap<String, Entry> state = new
ConcurrentHashMap<>();
- record Entry(int topologyEpoch, long currentDelayMs, long nextAttemptMs) {
}
+ record Entry(int topologyEpoch, int attempts, long nextAttemptMs) { }
public StreamsGroupTopologyDescriptionBackoff(Time time) {
+ this(time, new ExponentialBackoff(INITIAL_DELAY_MS, MULTIPLIER,
MAX_DELAY_MS, JITTER));
+ }
+
+ // Visible for testing
+ StreamsGroupTopologyDescriptionBackoff(Time time, ExponentialBackoff
exponentialBackoff) {
this.time = time;
+ this.exponentialBackoff = exponentialBackoff;
}
/**
@@ -55,55 +65,73 @@ public class StreamsGroupTopologyDescriptionBackoff {
/**
* Atomic check-and-arm. Returns true if no window was in effect and a new
one was
- * armed, false if a window was already active and nothing changed. Used
on the
- * heartbeat path to fold the "check + arm" pair into a single compute so
two
- * concurrent heartbeats for the same group cannot both arm the back-off,
and to
- * preserve the exponential chain when the previous window has expired
without a
- * push reaching the coordinator (e.g. the client never sent one, or the
push was
- * lost in flight before {@link #armOrExtend} could run).
+ * installed, false if a window was already active and nothing changed.
Used on the
+ * heartbeat path so two concurrent heartbeats for the same group cannot
both arm
+ * the back-off, and to preserve the exponential chain when the previous
window has
+ * expired without a push reaching the coordinator (e.g. the client never
sent one,
+ * or the push was lost in flight before {@link #armOrExtend} could run).
*/
public boolean armIfNotActive(String groupId, int topologyEpoch) {
- final long now = time.milliseconds();
- final boolean[] armed = new boolean[]{false};
- state.compute(groupId, (key, existing) -> {
+ while (true) {
+ final long now = time.milliseconds();
+ Entry existing = state.get(groupId);
if (existing != null
&& existing.topologyEpoch() == topologyEpoch
&& now < existing.nextAttemptMs()) {
- return existing;
+ return false;
}
- armed[0] = true;
- // Re-arm: continue the exponential chain at the same epoch (the
previous
- // window expired without a push completing); reset to
INITIAL_DELAY_MS on
- // an epoch advance, which implicitly drops the prior history.
- if (existing != null && existing.topologyEpoch() == topologyEpoch)
{
- long nextDelay = Math.min(existing.currentDelayMs() * 2,
MAX_DELAY_MS);
- return new Entry(topologyEpoch, nextDelay, now + nextDelay);
+ Entry next = computeNextEntry(existing, topologyEpoch, now);
+ boolean installed = existing == null
+ ? state.putIfAbsent(groupId, next) == null
+ : state.replace(groupId, existing, next);
+ if (installed) {
+ return true;
}
- return new Entry(topologyEpoch, INITIAL_DELAY_MS, now +
INITIAL_DELAY_MS);
- });
- return armed[0];
+ // Lost a race with a concurrent mutation; retry with the fresh
state.
+ }
}
/**
- * Arm a new back-off window or extend the existing one. If the existing
entry is for a
- * different topology epoch the window is reset to {@link
#INITIAL_DELAY_MS}.
+ * Arm a new back-off window or extend the existing one at the caller's
topology epoch.
+ *
+ * <p>Epoch-aware: if the stored entry is for a newer topology epoch than
{@code
+ * topologyEpoch} (a concurrent heartbeat already armed for the advanced
epoch while a
+ * late post-plugin callback finishes at the old epoch), the call is a
no-op so we do
+ * not overwrite the newer window. At the same epoch the attempt count
advances and
+ * the next delay is drawn from {@link ExponentialBackoff}; at a
stale-or-absent entry
+ * we start a fresh chain at {@code attempts=0}.
*/
public void armOrExtend(String groupId, int topologyEpoch) {
final long now = time.milliseconds();
+ state.compute(groupId, (key, existing) -> {
+ if (existing != null && existing.topologyEpoch() > topologyEpoch) {
+ return existing;
+ }
+ return computeNextEntry(existing, topologyEpoch, now);
+ });
+ }
+
+ /**
+ * Drop the back-off entry for a group at the given topology epoch.
Epoch-aware: only
+ * clears if the stored entry matches {@code topologyEpoch} (a late
post-plugin
+ * callback at the old epoch must not wipe a window a concurrent heartbeat
armed at
+ * the advanced epoch). Called on a successful push or a permanent plugin
failure.
+ */
+ public void clear(String groupId, int topologyEpoch) {
state.compute(groupId, (key, existing) -> {
if (existing == null || existing.topologyEpoch() != topologyEpoch)
{
- return new Entry(topologyEpoch, INITIAL_DELAY_MS, now +
INITIAL_DELAY_MS);
+ return existing;
}
- long nextDelay = Math.min(existing.currentDelayMs() * 2,
MAX_DELAY_MS);
- return new Entry(topologyEpoch, nextDelay, now + nextDelay);
+ return null;
});
}
/**
- * Drop the back-off entry for a group. Called on a successful push, a
permanent plugin
- * failure, or when the group is removed.
+ * Drop the back-off entry for a group unconditionally. Used by paths that
remove the
+ * group entirely (explicit {@code DeleteGroups}, periodic cleanup of
naturally-expired
+ * groups), where any in-flight back-off state is irrelevant because the
group is gone.
*/
- public void clear(String groupId) {
+ public void clearGroup(String groupId) {
state.remove(groupId);
}
@@ -111,4 +139,20 @@ public class StreamsGroupTopologyDescriptionBackoff {
Entry entry(String groupId) {
return state.get(groupId);
}
+
+ /**
+ * Build the next entry for an arm: same-epoch arms advance the attempt
count to
+ * continue the exponential chain; a different (or absent) epoch starts
fresh at
+ * {@code attempts=0}. The actual delay is drawn from {@link
ExponentialBackoff} so
+ * the multiplier / max / jitter are configured in one place rather than
hand-rolled
+ * twice.
+ */
+ private Entry computeNextEntry(Entry existing, int topologyEpoch, long
now) {
+ int nextAttempts =
+ (existing != null && existing.topologyEpoch() == topologyEpoch)
+ ? existing.attempts() + 1
+ : 0;
+ long delay = exponentialBackoff.backoff(nextAttempts);
+ return new Entry(topologyEpoch, nextAttempts, now + delay);
+ }
}
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionConverter.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionConverter.java
new file mode 100644
index 00000000000..94de07f38b1
--- /dev/null
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionConverter.java
@@ -0,0 +1,108 @@
+/*
+ * 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.coordinator.group.streams;
+
+import org.apache.kafka.common.errors.InvalidRequestException;
+import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateRequestData;
+import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription;
+import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription.GlobalStore;
+import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription.Node;
+import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription.Processor;
+import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription.Sink;
+import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription.Source;
+import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription.Subtopology;
+
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * Translates between the wire form of a streams topology description (as
carried on
+ * {@code StreamsGroupTopologyDescriptionUpdateRequest}) and the broker-side
+ * {@link StreamsGroupTopologyDescription} POJO that is handed to the plugin.
+ *
+ * <p>String collections are wrapped in {@link LinkedHashSet} so that the wire
ordering
+ * is preserved through to the POJO and any downstream pretty-printing.
+ */
+public final class StreamsGroupTopologyDescriptionConverter {
+
+ static final byte NODE_TYPE_SOURCE = 1;
+ static final byte NODE_TYPE_PROCESSOR = 2;
+ static final byte NODE_TYPE_SINK = 3;
+
+ private StreamsGroupTopologyDescriptionConverter() {
+ }
+
+ public static StreamsGroupTopologyDescription fromRequest(
+ StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription
wire
+ ) {
+ List<Subtopology> subtopologies = wire.subtopologies().stream()
+ .map(StreamsGroupTopologyDescriptionConverter::convertSubtopology)
+ .toList();
+ List<GlobalStore> globalStores = wire.globalStores().stream()
+ .map(StreamsGroupTopologyDescriptionConverter::convertGlobalStore)
+ .toList();
+ return new StreamsGroupTopologyDescription(subtopologies,
globalStores);
+ }
+
+ private static Subtopology convertSubtopology(
+
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionSubtopology
wire
+ ) {
+ List<Node> nodes = wire.nodes().stream()
+ .map(StreamsGroupTopologyDescriptionConverter::convertNode)
+ .toList();
+ return new Subtopology(wire.subtopologyId(), nodes);
+ }
+
+ private static GlobalStore convertGlobalStore(
+
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionGlobalStore
wire
+ ) {
+ Node source = convertNode(wire.source());
+ Node processor = convertNode(wire.processor());
+ if (!(source instanceof Source) || !(processor instanceof Processor)) {
+ throw new InvalidRequestException(
+ "Global store must be composed of a source and a processor
node."
+ );
+ }
+ return new GlobalStore((Source) source, (Processor) processor);
+ }
+
+ private static Node convertNode(
+
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode wire
+ ) {
+ return switch (wire.nodeType()) {
+ case NODE_TYPE_SOURCE -> new Source(
+ wire.name(),
+ new LinkedHashSet<>(wire.sourceTopics()),
+ new LinkedHashSet<>(wire.successors())
+ );
+ case NODE_TYPE_PROCESSOR -> new Processor(
+ wire.name(),
+ new LinkedHashSet<>(wire.stores()),
+ new LinkedHashSet<>(wire.successors())
+ );
+ case NODE_TYPE_SINK -> new Sink(
+ wire.name(),
+ Optional.ofNullable(wire.sinkTopic()),
+ new LinkedHashSet<>(wire.successors())
+ );
+ default -> throw new InvalidRequestException(
+ "Unknown topology node type: " + wire.nodeType()
+ );
+ };
+ }
+}
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 ca848518d10..bf6f62dc804 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
@@ -20,21 +20,28 @@ import
org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
import org.apache.kafka.common.protocol.Errors;
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.Objects;
import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
/**
- * Broker-level component that owns everything tied to the streams-group
topology
- * description plugin: the configured plugin reference, the per-group
re-solicitation
- * back-off, and the heartbeat-path gate that asks clients to push their
topology.
+ * Broker-level component that owns the streams-group topology description
plugin
+ * reference and the per-group re-solicitation back-off. The chain that drives
a push
+ * RPC (validate → convert → plugin → metadata write → back-off mutation)
lives on
+ * {@code GroupCoordinatorService}; this class exposes the building blocks
(plugin
+ * invocation, back-off mutations) and the heartbeat-path gate, but does not
assemble
+ * the chain itself.
*
* <p>This class is broker-level (one instance per {@code
GroupCoordinatorService}); the
* back-off map is keyed by {@code groupId} and shared across all partitions
hosted on the
* broker. State here is intentionally non-timeline and non-replayed: it is
rebuilt from
* scratch on broker restart, and the persisted {@code
StoredDescriptionTopologyEpoch} /
- * {@code FailedDescriptionTopologyEpoch} fields on each streams group drive
convergence
- * after a restart.
+ * {@code FailedDescriptionTopologyEpoch} fields on each streams group drive
+ * convergence after a restart.
*/
public class StreamsGroupTopologyDescriptionManager implements AutoCloseable {
private final Optional<StreamsGroupTopologyDescriptionPlugin> plugin;
@@ -114,6 +121,83 @@ public class StreamsGroupTopologyDescriptionManager
implements AutoCloseable {
return result;
}
+ /**
+ * Call {@code plugin.setTopology} and fold the result into a {@link
PluginOutcome}.
+ * The returned future never completes exceptionally — the outcome carries
the
+ * failure category so the caller can dispatch on it without try/catch on
the
+ * future. A synchronous throw from the plugin (which violates the SPI
contract) is
+ * mapped to a permanent failure with a generic message rather than
forwarding the
+ * raw exception text, and a {@code null} returned future is treated the
same way.
+ */
+ public CompletableFuture<PluginOutcome> invokeSetTopology(
+ String groupId,
+ int topologyEpoch,
+ StreamsGroupTopologyDescription description
+ ) {
+ if (plugin.isEmpty()) {
+ return CompletableFuture.completedFuture(
+ PluginOutcome.permanent("Topology description plugin
failed."));
+ }
+ final CompletableFuture<Void> pluginFuture;
+ try {
+ pluginFuture = Objects.requireNonNull(
+ plugin.get().setTopology(groupId, topologyEpoch, description));
+ } catch (Exception e) {
+ // A synchronous throw violates the SPI contract — implementations
must signal
+ // failures by completing the future exceptionally. Treat it as a
permanent
+ // failure with a stable, generic client-visible message so we
don't forward
+ // an unbounded or null exception message that could leak plugin
internals.
+ return CompletableFuture.completedFuture(
+ PluginOutcome.permanent("Topology description plugin
failed."));
+ }
+ return pluginFuture.handle((unused, throwable) -> {
+ if (throwable == null) {
+ return PluginOutcome.success();
+ }
+ // CompletionException / ExecutionException can legally carry a
null cause; if a
+ // plugin completes its future with one of those (rare but legal),
+ // maybeUnwrapException returns null. Treat that as a transient
failure with a
+ // generic message rather than NPE-ing inside this handle and
losing the
+ // transient/permanent classification downstream.
+ Throwable cause = Errors.maybeUnwrapException(throwable);
+ if (cause == null) {
+ return PluginOutcome.transientFailure("Plugin failure (no
cause).");
+ }
+ if (cause instanceof
StreamsTopologyDescriptionPermanentFailureException) {
+ return PluginOutcome.permanent(cause.getMessage());
+ }
+ return PluginOutcome.transientFailure(cause.getMessage());
+ });
+ }
+
+ /**
+ * Arm or extend the back-off window for a group at the given topology
epoch.
+ * Delegates to {@link StreamsGroupTopologyDescriptionBackoff#armOrExtend}.
+ */
+ public void armBackoff(String groupId, int topologyEpoch) {
+ backoff.armOrExtend(groupId, topologyEpoch);
+ }
+
+ /**
+ * Drop the back-off entry for a group at the given topology epoch.
Epoch-scoped so a
+ * late post-plugin callback at an old epoch cannot wipe a window a
concurrent
+ * heartbeat armed at the advanced epoch. Delegates to
+ * {@link StreamsGroupTopologyDescriptionBackoff#clear}.
+ */
+ public void clearBackoff(String groupId, int topologyEpoch) {
+ backoff.clear(groupId, topologyEpoch);
+ }
+
+ /**
+ * Drop the back-off entry for a group unconditionally. Used by paths that
remove the
+ * group entirely (explicit DeleteGroups, periodic cleanup of
naturally-expired
+ * groups, post-plugin write failing with GroupIdNotFoundException).
Delegates to
+ * {@link StreamsGroupTopologyDescriptionBackoff#clearGroup}.
+ */
+ public void clearBackoffGroup(String groupId) {
+ backoff.clearGroup(groupId);
+ }
+
// Visible for testing.
StreamsGroupTopologyDescriptionBackoff backoff() {
return backoff;
@@ -126,4 +210,25 @@ public class StreamsGroupTopologyDescriptionManager
implements AutoCloseable {
byte staleCode = Status.STALE_TOPOLOGY.code();
return response.status().stream().anyMatch(s -> s.statusCode() ==
staleCode);
}
+
+ /**
+ * Outcome of a {@code plugin.setTopology} call, folded into a value so
the caller can
+ * dispatch on {@link Kind} without try/catch on the underlying future.
+ */
+ public record PluginOutcome(Kind kind, String message) {
+
+ public enum Kind { SUCCESS, PERMANENT, TRANSIENT }
+
+ public static PluginOutcome success() {
+ return new PluginOutcome(Kind.SUCCESS, null);
+ }
+
+ public static PluginOutcome permanent(String message) {
+ return new PluginOutcome(Kind.PERMANENT, message);
+ }
+
+ public static PluginOutcome transientFailure(String message) {
+ return new PluginOutcome(Kind.TRANSIENT, message);
+ }
+ }
}
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 7ed02b37d92..da4c4ae6591 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
@@ -18,15 +18,21 @@ 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.UnknownMemberIdException;
import org.apache.kafka.common.internals.Topic;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
+import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateRequestData;
+import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateResponseData;
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.LogContext;
import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
import org.apache.kafka.coordinator.common.runtime.CoordinatorRuntime;
import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescriptionPlugin;
+import
org.apache.kafka.coordinator.group.api.streams.StreamsTopologyDescriptionPermanentFailureException;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics;
import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
import org.apache.kafka.server.share.persister.NoOpStatePersister;
@@ -34,28 +40,35 @@ import org.apache.kafka.server.util.timer.MockTimer;
import org.junit.jupiter.api.Test;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.TimeUnit;
import static
org.apache.kafka.common.requests.StreamsGroupHeartbeatResponse.Status;
import static
org.apache.kafka.coordinator.common.runtime.TestUtil.requestContext;
import static
org.apache.kafka.coordinator.group.GroupConfigManagerTest.createConfigManager;
+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.assertTrue;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
- * Tests for the heartbeat post-processing on {@link GroupCoordinatorService}
that sets
- * {@code TopologyDescriptionRequired} on the response when a topology
description plugin
- * is configured and the stored/failed epochs lag the current topology epoch.
+ * Tests for the topology-description plugin paths added to {@link
GroupCoordinatorService}:
+ * the new {@code streamsGroupTopologyDescriptionUpdate} RPC, the heartbeat
post-processing
+ * that sets {@code TopologyDescriptionRequired}, and the back-off interaction.
*/
public class GroupCoordinatorServiceTopologyDescriptionTest {
@@ -99,6 +112,324 @@ public class
GroupCoordinatorServiceTopologyDescriptionTest {
return service;
}
+ @Test
+ public void testUpdateRejectedWhenCoordinatorNotActive() throws Exception {
+ GroupCoordinatorService service = buildService(mockRuntime(),
Optional.empty(), false);
+
+ StreamsGroupTopologyDescriptionUpdateResponseData response =
service.streamsGroupTopologyDescriptionUpdate(
+ requestContext(ApiKeys.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_UPDATE),
+ validUpdateRequest()
+ ).get(5, TimeUnit.SECONDS);
+
+ assertEquals(Errors.COORDINATOR_NOT_AVAILABLE.code(),
response.errorCode());
+ }
+
+ @Test
+ public void testUpdateReturnsUnsupportedVersionWhenNoPlugin() throws
Exception {
+ GroupCoordinatorService service = buildService(mockRuntime(),
Optional.empty(), true);
+
+ StreamsGroupTopologyDescriptionUpdateResponseData response =
service.streamsGroupTopologyDescriptionUpdate(
+ requestContext(ApiKeys.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_UPDATE),
+ validUpdateRequest()
+ ).get(5, TimeUnit.SECONDS);
+
+ assertEquals(Errors.UNSUPPORTED_VERSION.code(), response.errorCode());
+ assertNotNull(response.errorMessage());
+ }
+
+ @Test
+ public void testUpdateRejectsEmptyMemberId() throws Exception {
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ GroupCoordinatorService service = buildService(mockRuntime(),
Optional.of(plugin), true);
+
+ StreamsGroupTopologyDescriptionUpdateResponseData response =
service.streamsGroupTopologyDescriptionUpdate(
+ requestContext(ApiKeys.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_UPDATE),
+ validUpdateRequest().setMemberId("")
+ ).get(5, TimeUnit.SECONDS);
+
+ assertEquals(Errors.INVALID_REQUEST.code(), response.errorCode());
+ assertEquals("MemberId can't be empty.", response.errorMessage());
+ }
+
+ @Test
+ public void testUpdateRejectsEmptyGroupId() throws Exception {
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ GroupCoordinatorService service = buildService(mockRuntime(),
Optional.of(plugin), true);
+
+ StreamsGroupTopologyDescriptionUpdateResponseData response =
service.streamsGroupTopologyDescriptionUpdate(
+ requestContext(ApiKeys.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_UPDATE),
+ validUpdateRequest().setGroupId("")
+ ).get(5, TimeUnit.SECONDS);
+
+ assertEquals(Errors.INVALID_REQUEST.code(), response.errorCode());
+ }
+
+ @Test
+ public void testUpdateSuccessPersistsStoredEpoch() throws Exception {
+ 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.completedFuture(null));
+
+ 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.NONE.code(), response.errorCode());
+ verify(plugin, times(1)).setTopology(eq("foo"), eq(3), any());
+ verify(runtime, times(1)).scheduleWriteOperation(
+ eq("streams-group-set-stored-topology-epoch"), eq(GROUP_TP),
any());
+
+ // Back-off must be cleared on success: a subsequent heartbeat at the
same epoch
+ // (stored still lags in this mock-only world because the
metadata-record write
+ // is captured but not replayed) should set the flag again.
+ assertTrue(heartbeatTopologyDescriptionRequired(runtime, service, 3,
-1, -1));
+ }
+
+ @Test
+ public void testUpdatePermanentFailurePersistsFailedEpoch() throws
Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(plugin.setTopology(anyString(), anyInt(), any()))
+ .thenReturn(CompletableFuture.failedFuture(
+ new StreamsTopologyDescriptionPermanentFailureException("too
large")));
+ when(runtime.scheduleReadOperation(
+ eq("streams-group-topology-description-validate"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(null));
+ when(runtime.scheduleWriteOperation(
+ eq("streams-group-set-failed-topology-epoch"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(null));
+
+ 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.STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED.code(),
response.errorCode());
+ assertEquals("too large", response.errorMessage());
+ verify(runtime, times(1)).scheduleWriteOperation(
+ eq("streams-group-set-failed-topology-epoch"), eq(GROUP_TP),
any());
+ }
+
+ @Test
+ public void testUpdateTransientFailureWritesNoRecord() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(plugin.setTopology(anyString(), anyInt(), any()))
+ .thenReturn(CompletableFuture.failedFuture(new
RuntimeException("backend offline")));
+ when(runtime.scheduleReadOperation(
+ eq("streams-group-topology-description-validate"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(null));
+
+ 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.STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED.code(),
response.errorCode());
+ assertEquals("backend offline", response.errorMessage());
+ verify(runtime, never()).scheduleWriteOperation(
+ eq("streams-group-set-stored-topology-epoch"), any(), any());
+ verify(runtime, never()).scheduleWriteOperation(
+ eq("streams-group-set-failed-topology-epoch"), any(), any());
+
+ // Back-off must be armed on transient failure: a subsequent heartbeat
at the same
+ // epoch is suppressed rather than re-soliciting immediately.
+ assertFalse(heartbeatTopologyDescriptionRequired(runtime, service, 3,
-1, -1));
+ }
+
+ @Test
+ public void testUpdatePluginFutureWithNullCauseIsTreatedAsTransient()
throws Exception {
+ // CompletionException / ExecutionException can legally carry a null
cause. If a
+ // plugin completes its future with one of those (rare but legal), the
handle()
+ // callback must not NPE on cause.getMessage() — that would lose the
+ // transient/permanent classification and surface
UNKNOWN_SERVER_ERROR. The null
+ // cause is treated as a transient failure with a generic message.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(plugin.setTopology(anyString(), anyInt(), any()))
+ .thenReturn(CompletableFuture.failedFuture(new
CompletionException(null)));
+ when(runtime.scheduleReadOperation(
+ eq("streams-group-topology-description-validate"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(null));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+
+ StreamsGroupTopologyDescriptionUpdateResponseData response =
service.streamsGroupTopologyDescriptionUpdate(
+ requestContext(ApiKeys.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_UPDATE),
+ validUpdateRequest()
+ ).get(5, TimeUnit.SECONDS);
+
+ // Treated as transient: STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED
rather than
+ // UNKNOWN_SERVER_ERROR; no metadata record written; back-off armed.
+ assertEquals(Errors.STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED.code(),
response.errorCode());
+ assertNotNull(response.errorMessage());
+ verify(runtime, never()).scheduleWriteOperation(
+ eq("streams-group-set-stored-topology-epoch"), any(), any());
+ verify(runtime, never()).scheduleWriteOperation(
+ eq("streams-group-set-failed-topology-epoch"), any(), any());
+ assertFalse(heartbeatTopologyDescriptionRequired(runtime, service, 3,
-1, -1));
+ }
+
+ @Test
+ public void
testUpdateBackoffArmedWhenStoredEpochWriteFailsAfterPluginSuccess() throws
Exception {
+ 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
RuntimeException("write failed")));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+
+ StreamsGroupTopologyDescriptionUpdateResponseData response =
service.streamsGroupTopologyDescriptionUpdate(
+ requestContext(ApiKeys.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_UPDATE),
+ validUpdateRequest()
+ ).get(5, TimeUnit.SECONDS);
+
+ // The exceptionally branch translates the write failure into an error
response.
+ assertEquals(Errors.UNKNOWN_SERVER_ERROR.code(), response.errorCode());
+ verify(plugin, times(1)).setTopology(eq("foo"), eq(3), any());
+
+ // Plugin succeeded, write failed. BackoffAction defaulted to ARM and
the thenApply
+ // that would have set CLEAR never ran, so whenComplete armed the
back-off.
+ assertFalse(heartbeatTopologyDescriptionRequired(runtime, service, 3,
-1, -1));
+ }
+
+ @Test
+ public void testUpdatePreValidationFailureDoesNotArmBackoff() throws
Exception {
+ // A fenced/stale member (or, once routing lands, an unauthorized
caller) whose
+ // validateStreamsGroupMember check fails must not arm the per-group
back-off:
+ // legitimate members of the same group must still get
TopologyDescriptionRequired
+ // on their next heartbeat at the same epoch.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(runtime.scheduleReadOperation(
+ eq("streams-group-topology-description-validate"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.failedFuture(
+ new UnknownMemberIdException("Member fenced from the group.")));
+
+ 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.UNKNOWN_MEMBER_ID.code(), response.errorCode());
+ verify(plugin, never()).setTopology(anyString(), anyInt(), any());
+
+ // Pre-plugin failure must leave the back-off untouched. A legitimate
heartbeat at
+ // the same epoch must still get the flag — otherwise a
fenced/unauthorized caller
+ // could grief the entire group's re-solicitation until the back-off
window expires.
+ 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
+ // write for StoredDescriptionTopologyEpoch fails with
GroupIdNotFoundException. The
+ // push has already taken effect at the plugin and the group is gone,
so the back-off
+ // must be dropped rather than re-armed for a now-orphan group.
+ 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 GroupIdNotFoundException("Group deleted 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.GROUP_ID_NOT_FOUND.code(), response.errorCode());
+ verify(plugin, times(1)).setTopology(eq("foo"), eq(3), any());
+
+ // Back-off must be dropped, not armed: nobody will ever clear it for
a deleted group.
+ // A subsequent heartbeat at the same epoch (if the group somehow
comes back) must
+ // still get the flag — i.e. the orphan back-off entry is gone.
+ assertTrue(heartbeatTopologyDescriptionRequired(runtime, service, 3,
-1, -1));
+ }
+
+ @Test
+ public void testUpdatePluginReturnsNullFutureIsTreatedAsPermanentFailure()
throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(plugin.setTopology(anyString(), anyInt(),
any())).thenReturn(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-failed-topology-epoch"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(null));
+
+ 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.STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED.code(),
response.errorCode());
+ assertEquals("Topology description plugin failed.",
response.errorMessage());
+ // Treated as permanent failure: FailedDescriptionTopologyEpoch is
written.
+ verify(runtime, times(1)).scheduleWriteOperation(
+ eq("streams-group-set-failed-topology-epoch"), eq(GROUP_TP),
any());
+ verify(runtime, never()).scheduleWriteOperation(
+ eq("streams-group-set-stored-topology-epoch"), any(), any());
+ }
+
@Test
public void testHeartbeatSetsTopologyDescriptionRequiredWhenStoredLags()
throws Exception {
CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
@@ -226,6 +557,38 @@ public class
GroupCoordinatorServiceTopologyDescriptionTest {
assertFalse(result.data().topologyDescriptionRequired());
}
+ @Test
+ public void testHeartbeatDecorationFailurePreservesCommittedResponse()
throws Exception {
+ // If maybeSetTopologyDescriptionRequired throws while decorating an
+ // already-committed successful heartbeat (for example, because the
response carries
+ // an unexpected shape such as a null Status element), the service
must NOT translate
+ // that into an error response — the broker-side state change has
already happened,
+ // so we return the committed result as-is and let the next heartbeat
retry.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+
+ // Construct a heartbeat result where responseHasStaleTopology will
NPE on the null
+ // status element. errorCode is NONE so we exercise the success path.
+ StreamsGroupHeartbeatResponseData response = new
StreamsGroupHeartbeatResponseData()
+ .setStatus(Collections.singletonList(null));
+ when(runtime.scheduleWriteOperation(
+ eq("streams-group-heartbeat"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(
+ new StreamsGroupHeartbeatResult(response, Map.of(), 5, -1, -1)));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ StreamsGroupHeartbeatResult result = service.streamsGroupHeartbeat(
+ requestContext(ApiKeys.STREAMS_GROUP_HEARTBEAT),
validHeartbeatRequest()
+ ).get(5, TimeUnit.SECONDS);
+
+ // No error translation: the response carries NONE and the original
status, the flag
+ // is left unset because decoration could not complete.
+ assertEquals(Errors.NONE.code(), result.data().errorCode());
+ assertFalse(result.data().topologyDescriptionRequired());
+ }
+
@Test
public void testHeartbeatNeverSetsFlagWithoutPlugin() throws Exception {
CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
@@ -271,6 +634,17 @@ public class
GroupCoordinatorServiceTopologyDescriptionTest {
verify(plugin, times(1)).close();
}
+ private static StreamsGroupTopologyDescriptionUpdateRequestData
validUpdateRequest() {
+ return new StreamsGroupTopologyDescriptionUpdateRequestData()
+ .setGroupId("foo")
+ .setMemberId(Uuid.randomUuid().toString())
+ .setTopologyEpoch(3)
+ .setTopologyDescription(
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription()
+ .setSubtopologies(List.of())
+ .setGlobalStores(List.of()));
+ }
+
private static StreamsGroupHeartbeatRequestData validHeartbeatRequest() {
return new StreamsGroupHeartbeatRequestData()
.setGroupId("foo")
@@ -282,4 +656,35 @@ public class
GroupCoordinatorServiceTopologyDescriptionTest {
.setWarmupTasks(List.of())
.setTopology(new StreamsGroupHeartbeatRequestData.Topology());
}
+
+ /**
+ * Drive a heartbeat through the service after an update test and report
whether the
+ * topology-description-required flag was set on the response. Used to
observe the
+ * back-off state behaviourally: the flag is set iff the back-off window
is not in
+ * effect for the given epoch, so the assertion stands in for "back-off
cleared" vs
+ * "back-off armed".
+ */
+ private static boolean heartbeatTopologyDescriptionRequired(
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime,
+ GroupCoordinatorService service,
+ int currentEpoch,
+ int storedEpoch,
+ int failedEpoch
+ ) throws Exception {
+ when(runtime.scheduleWriteOperation(
+ eq("streams-group-heartbeat"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(new
StreamsGroupHeartbeatResult(
+ new StreamsGroupHeartbeatResponseData(),
+ Map.of(),
+ currentEpoch,
+ storedEpoch,
+ failedEpoch
+ )));
+ StreamsGroupHeartbeatResult result = service.streamsGroupHeartbeat(
+ requestContext(ApiKeys.STREAMS_GROUP_HEARTBEAT),
validHeartbeatRequest()
+ ).get(5, TimeUnit.SECONDS);
+ return result.data().topologyDescriptionRequired();
+ }
}
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java
index 448145c54fa..5f375d67c2e 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java
@@ -19,6 +19,7 @@ package org.apache.kafka.coordinator.group;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.GroupIdNotFoundException;
import org.apache.kafka.common.errors.GroupNotEmptyException;
+import org.apache.kafka.common.errors.InvalidRequestException;
import org.apache.kafka.common.errors.UnknownMemberIdException;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData;
@@ -244,21 +245,27 @@ public class GroupCoordinatorShardTest {
);
// Happy path: manager returns a member, shard returns void.
- when(groupMetadataManager.validateStreamsGroupMember("foo", "m1",
100L))
+
when(groupMetadataManager.validateStreamsGroupTopologyDescriptionUpdate("foo",
"m1", 3, 100L))
.thenReturn(mock(StreamsGroupMember.class));
- coordinator.validateStreamsGroupMember("foo", "m1", 100L);
+ coordinator.validateStreamsGroupTopologyDescriptionUpdate("foo", "m1",
3, 100L);
// GROUP_ID_NOT_FOUND propagates.
- when(groupMetadataManager.validateStreamsGroupMember("missing", "m1",
100L))
+
when(groupMetadataManager.validateStreamsGroupTopologyDescriptionUpdate("missing",
"m1", 3, 100L))
.thenThrow(new GroupIdNotFoundException("nope"));
assertThrows(GroupIdNotFoundException.class,
- () -> coordinator.validateStreamsGroupMember("missing", "m1",
100L));
+ () ->
coordinator.validateStreamsGroupTopologyDescriptionUpdate("missing", "m1", 3,
100L));
// UNKNOWN_MEMBER_ID propagates.
- when(groupMetadataManager.validateStreamsGroupMember("foo",
"stranger", 100L))
+
when(groupMetadataManager.validateStreamsGroupTopologyDescriptionUpdate("foo",
"stranger", 3, 100L))
.thenThrow(new UnknownMemberIdException("not a member"));
assertThrows(UnknownMemberIdException.class,
- () -> coordinator.validateStreamsGroupMember("foo", "stranger",
100L));
+ () ->
coordinator.validateStreamsGroupTopologyDescriptionUpdate("foo", "stranger", 3,
100L));
+
+ // Stale topology epoch propagates as INVALID_REQUEST.
+
when(groupMetadataManager.validateStreamsGroupTopologyDescriptionUpdate("foo",
"m1", 2, 100L))
+ .thenThrow(new InvalidRequestException("stale epoch"));
+ assertThrows(InvalidRequestException.class,
+ () ->
coordinator.validateStreamsGroupTopologyDescriptionUpdate("foo", "m1", 2,
100L));
}
@Test
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
index f6e4033d9aa..7c033026b08 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
@@ -10820,19 +10820,19 @@ public class GroupMetadataManagerTest {
}
@Test
- public void testValidateStreamsGroupMemberThrowsWhenGroupAbsent() {
- // KIP-1331: validateStreamsGroupMember surfaces GROUP_ID_NOT_FOUND
for the upcoming
- // StreamsGroupTopologyDescriptionUpdate handler.
+ public void
testValidateStreamsGroupTopologyDescriptionUpdateThrowsWhenGroupAbsent() {
+ // KIP-1331: validateStreamsGroupTopologyDescriptionUpdate surfaces
GROUP_ID_NOT_FOUND for
+ // the StreamsGroupTopologyDescriptionUpdate handler.
GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder().build();
assertThrows(GroupIdNotFoundException.class,
- () -> context.groupMetadataManager.validateStreamsGroupMember(
- "nonexistent", "m1", context.lastCommittedOffset));
+ () ->
context.groupMetadataManager.validateStreamsGroupTopologyDescriptionUpdate(
+ "nonexistent", "m1", 0, context.lastCommittedOffset));
}
@Test
- public void testValidateStreamsGroupMemberThrowsWhenMemberAbsent() {
- // KIP-1331: validateStreamsGroupMember surfaces UNKNOWN_MEMBER_ID for
the upcoming
- // StreamsGroupTopologyDescriptionUpdate handler.
+ public void
testValidateStreamsGroupTopologyDescriptionUpdateThrowsWhenMemberAbsent() {
+ // KIP-1331: validateStreamsGroupTopologyDescriptionUpdate surfaces
UNKNOWN_MEMBER_ID for
+ // the StreamsGroupTopologyDescriptionUpdate handler.
String groupId = "streams-group";
GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder().build();
// Replaying a metadata record materializes a streams group with no
members; commit so the
@@ -10842,14 +10842,38 @@ public class GroupMetadataManagerTest {
context.commit();
assertThrows(UnknownMemberIdException.class,
- () -> context.groupMetadataManager.validateStreamsGroupMember(
- groupId, "stranger", context.lastCommittedOffset));
+ () ->
context.groupMetadataManager.validateStreamsGroupTopologyDescriptionUpdate(
+ groupId, "stranger", -1, context.lastCommittedOffset));
}
@Test
- public void testValidateStreamsGroupMemberDoesNotSeeUncommittedFence() {
- // KIP-1331: validateStreamsGroupMember reads at committedOffset, so
an uncommitted fence
- // (member tombstone) must not make a still-committed member appear
unknown.
+ public void
testValidateStreamsGroupTopologyDescriptionUpdateRejectsStaleEpoch() {
+ // KIP-1331: validateStreamsGroupTopologyDescriptionUpdate must reject
a pushedEpoch that
+ // does not match the group's current topology epoch with
INVALID_REQUEST. Without this
+ // check, a stale push could regress storedDescriptionTopologyEpoch
and the heartbeat
+ // gate would never converge.
+ String groupId = "streams-group";
+ String memberId = "m1";
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder().build();
+
+ StreamsGroupMember member =
streamsGroupMemberBuilderWithDefaults(memberId).build();
+
context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord(
+ groupId, 1, 0L, -1, Map.of(), -1, -1));
+
context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord(groupId,
member));
+
context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId,
member));
+ context.commit();
+
+ // No topology has been pushed via heartbeat, so
currentTopologyEpoch() returns -1.
+ // A push claiming epoch 4 must be rejected.
+ assertThrows(InvalidRequestException.class,
+ () ->
context.groupMetadataManager.validateStreamsGroupTopologyDescriptionUpdate(
+ groupId, memberId, 4, context.lastCommittedOffset));
+ }
+
+ @Test
+ public void
testValidateStreamsGroupTopologyDescriptionUpdateDoesNotSeeUncommittedFence() {
+ // KIP-1331: validateStreamsGroupTopologyDescriptionUpdate reads at
committedOffset, so an
+ // uncommitted fence (member tombstone) must not make a
still-committed member appear unknown.
String groupId = "streams-group";
String memberId = "m1";
GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder().build();
@@ -10870,13 +10894,15 @@ public class GroupMetadataManagerTest {
context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord(groupId,
memberId));
// Validating at the still-committed offset must succeed; the
uncommitted tombstone is invisible.
- StreamsGroupMember resolved =
context.groupMetadataManager.validateStreamsGroupMember(
- groupId, memberId, committedWithMember);
+ // currentTopologyEpoch() is -1 since no topology has been pushed via
heartbeat.
+ StreamsGroupMember resolved =
context.groupMetadataManager.validateStreamsGroupTopologyDescriptionUpdate(
+ groupId, memberId, -1, committedWithMember);
assertEquals(memberId, resolved.memberId());
// Latest in-memory state, by contrast, sees the tombstone — verify by
querying with Long.MAX_VALUE.
assertThrows(UnknownMemberIdException.class,
- () ->
context.groupMetadataManager.validateStreamsGroupMember(groupId, memberId,
Long.MAX_VALUE));
+ () ->
context.groupMetadataManager.validateStreamsGroupTopologyDescriptionUpdate(
+ groupId, memberId, -1, Long.MAX_VALUE));
}
@Test
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/api/streams/StreamsGroupTopologyDescriptionTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/api/streams/StreamsGroupTopologyDescriptionTest.java
index 1838ff96af6..dc842bccc34 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/api/streams/StreamsGroupTopologyDescriptionTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/api/streams/StreamsGroupTopologyDescriptionTest.java
@@ -81,14 +81,11 @@ public class StreamsGroupTopologyDescriptionTest {
}
@Test
- public void testCollectionsAreDefensivelyCopied() {
- Set<String> mutableTopics = new HashSet<>(Set.of("in"));
+ public void testCollectionAccessorsAreUnmodifiable() {
StreamsGroupTopologyDescription.Source src = new
StreamsGroupTopologyDescription.Source(
- "src", mutableTopics, Set.of("proc"));
- mutableTopics.add("rogue");
-
- assertEquals(Set.of("in"), src.topics());
+ "src", new HashSet<>(Set.of("in")), Set.of("proc"));
assertThrows(UnsupportedOperationException.class, () ->
src.topics().add("nope"));
+ assertThrows(UnsupportedOperationException.class, () ->
src.successors().add("nope"));
}
@Test
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionBackoffTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionBackoffTest.java
index d6eae64a3e8..3a32d7769e3 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionBackoffTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionBackoffTest.java
@@ -17,6 +17,7 @@
package org.apache.kafka.coordinator.group.streams;
import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.internals.ExponentialBackoff;
import org.junit.jupiter.api.Test;
@@ -28,12 +29,27 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class StreamsGroupTopologyDescriptionBackoffTest {
+ /**
+ * Build a back-off backed by a jitter-free {@link ExponentialBackoff} so
delay
+ * assertions are deterministic.
+ */
+ private static StreamsGroupTopologyDescriptionBackoff
deterministicBackoff(MockTime time) {
+ return new StreamsGroupTopologyDescriptionBackoff(
+ time,
+ new ExponentialBackoff(
+ StreamsGroupTopologyDescriptionBackoff.INITIAL_DELAY_MS,
+ StreamsGroupTopologyDescriptionBackoff.MULTIPLIER,
+ StreamsGroupTopologyDescriptionBackoff.MAX_DELAY_MS,
+ 0.0));
+ }
+
@Test
public void testFirstArmUsesInitialDelay() {
MockTime time = new MockTime();
- StreamsGroupTopologyDescriptionBackoff backoff = new
StreamsGroupTopologyDescriptionBackoff(time);
+ StreamsGroupTopologyDescriptionBackoff backoff =
deterministicBackoff(time);
backoff.armOrExtend("g", 1);
assertTrue(backoff.isActive("g", 1));
+ assertEquals(0, backoff.entry("g").attempts());
time.sleep(StreamsGroupTopologyDescriptionBackoff.INITIAL_DELAY_MS -
1);
assertTrue(backoff.isActive("g", 1));
time.sleep(1);
@@ -41,43 +57,42 @@ public class StreamsGroupTopologyDescriptionBackoffTest {
}
@Test
- public void testConsecutiveArmsDoubleTheWindowUpToMax() {
+ public void testConsecutiveArmsAdvanceAttemptsUpToCap() {
MockTime time = new MockTime();
- StreamsGroupTopologyDescriptionBackoff backoff = new
StreamsGroupTopologyDescriptionBackoff(time);
+ StreamsGroupTopologyDescriptionBackoff backoff =
deterministicBackoff(time);
- long expected =
StreamsGroupTopologyDescriptionBackoff.INITIAL_DELAY_MS;
backoff.armOrExtend("g", 1);
- assertEquals(expected, backoff.entry("g").currentDelayMs());
+ assertEquals(0, backoff.entry("g").attempts());
- // Each consecutive arm at the same epoch doubles, until we hit the
cap.
+ // Each consecutive arm at the same epoch advances the attempt count.
The actual
+ // delay is delegated to ExponentialBackoff and capped at MAX_DELAY_MS.
for (int i = 0; i < 20; i++) {
backoff.armOrExtend("g", 1);
- expected = Math.min(expected * 2,
StreamsGroupTopologyDescriptionBackoff.MAX_DELAY_MS);
- assertEquals(expected, backoff.entry("g").currentDelayMs(),
- "iteration " + i);
+ assertEquals(i + 1, backoff.entry("g").attempts(), "iteration " +
i);
}
- assertEquals(StreamsGroupTopologyDescriptionBackoff.MAX_DELAY_MS,
- backoff.entry("g").currentDelayMs());
+
+ // Window length is capped: arming when the formula would exceed
MAX_DELAY_MS still
+ // produces a window no longer than MAX_DELAY_MS.
+ long delay = backoff.entry("g").nextAttemptMs() - time.milliseconds();
+ assertEquals(StreamsGroupTopologyDescriptionBackoff.MAX_DELAY_MS,
delay);
}
@Test
- public void testDifferentEpochResetsTheWindow() {
+ public void testDifferentEpochResetsAttempts() {
MockTime time = new MockTime();
- StreamsGroupTopologyDescriptionBackoff backoff = new
StreamsGroupTopologyDescriptionBackoff(time);
+ StreamsGroupTopologyDescriptionBackoff backoff =
deterministicBackoff(time);
backoff.armOrExtend("g", 1);
- backoff.armOrExtend("g", 1); // doubled
- long doubled = backoff.entry("g").currentDelayMs();
- assertTrue(doubled >
StreamsGroupTopologyDescriptionBackoff.INITIAL_DELAY_MS);
+ backoff.armOrExtend("g", 1); // advances attempts
+ assertEquals(1, backoff.entry("g").attempts());
backoff.armOrExtend("g", 2);
- assertEquals(StreamsGroupTopologyDescriptionBackoff.INITIAL_DELAY_MS,
- backoff.entry("g").currentDelayMs());
+ assertEquals(0, backoff.entry("g").attempts());
assertEquals(2, backoff.entry("g").topologyEpoch());
}
@Test
public void testIsActiveIsScopedToEpoch() {
MockTime time = new MockTime();
- StreamsGroupTopologyDescriptionBackoff backoff = new
StreamsGroupTopologyDescriptionBackoff(time);
+ StreamsGroupTopologyDescriptionBackoff backoff =
deterministicBackoff(time);
backoff.armOrExtend("g", 1);
assertTrue(backoff.isActive("g", 1));
// A query at a different epoch never matches — the broker should
re-solicit.
@@ -85,30 +100,70 @@ public class StreamsGroupTopologyDescriptionBackoffTest {
}
@Test
- public void testClearRemovesEntry() {
+ public void testClearRemovesEntryWhenEpochMatches() {
MockTime time = new MockTime();
- StreamsGroupTopologyDescriptionBackoff backoff = new
StreamsGroupTopologyDescriptionBackoff(time);
+ StreamsGroupTopologyDescriptionBackoff backoff =
deterministicBackoff(time);
backoff.armOrExtend("g", 1);
assertNotNull(backoff.entry("g"));
- backoff.clear("g");
+ backoff.clear("g", 1);
assertNull(backoff.entry("g"));
assertFalse(backoff.isActive("g", 1));
}
+ @Test
+ public void testClearIsNoOpWhenEpochDoesNotMatch() {
+ // A late post-plugin callback at the old epoch must not wipe a window
a concurrent
+ // heartbeat armed at the advanced epoch.
+ MockTime time = new MockTime();
+ StreamsGroupTopologyDescriptionBackoff backoff =
deterministicBackoff(time);
+ backoff.armOrExtend("g", 6);
+ long armedAt = backoff.entry("g").nextAttemptMs();
+
+ backoff.clear("g", 5);
+ assertNotNull(backoff.entry("g"));
+ assertEquals(6, backoff.entry("g").topologyEpoch());
+ assertEquals(armedAt, backoff.entry("g").nextAttemptMs());
+ }
+
+ @Test
+ public void testClearGroupRemovesEntryUnconditionally() {
+ // Used by paths that remove the group entirely (DeleteGroups,
periodic cleanup):
+ // back-off state is irrelevant because the group is gone.
+ MockTime time = new MockTime();
+ StreamsGroupTopologyDescriptionBackoff backoff =
deterministicBackoff(time);
+ backoff.armOrExtend("g", 6);
+ backoff.clearGroup("g");
+ assertNull(backoff.entry("g"));
+ }
+
+ @Test
+ public void testArmOrExtendIsNoOpWhenStoredEpochIsNewer() {
+ // A late post-plugin callback at the old epoch must not overwrite an
entry a
+ // concurrent heartbeat armed at the advanced epoch.
+ MockTime time = new MockTime();
+ StreamsGroupTopologyDescriptionBackoff backoff =
deterministicBackoff(time);
+ backoff.armOrExtend("g", 6);
+ long armedAt = backoff.entry("g").nextAttemptMs();
+
+ backoff.armOrExtend("g", 5);
+ assertEquals(6, backoff.entry("g").topologyEpoch());
+ assertEquals(0, backoff.entry("g").attempts());
+ assertEquals(armedAt, backoff.entry("g").nextAttemptMs());
+ }
+
@Test
public void testArmIfNotActiveArmsWhenIdle() {
MockTime time = new MockTime();
- StreamsGroupTopologyDescriptionBackoff backoff = new
StreamsGroupTopologyDescriptionBackoff(time);
+ StreamsGroupTopologyDescriptionBackoff backoff =
deterministicBackoff(time);
assertTrue(backoff.armIfNotActive("g", 1));
assertTrue(backoff.isActive("g", 1));
- assertEquals(StreamsGroupTopologyDescriptionBackoff.INITIAL_DELAY_MS,
- backoff.entry("g").currentDelayMs());
+ assertEquals(0, backoff.entry("g").attempts());
}
@Test
public void testArmIfNotActiveReturnsFalseWhenAlreadyActive() {
MockTime time = new MockTime();
- StreamsGroupTopologyDescriptionBackoff backoff = new
StreamsGroupTopologyDescriptionBackoff(time);
+ StreamsGroupTopologyDescriptionBackoff backoff =
deterministicBackoff(time);
assertTrue(backoff.armIfNotActive("g", 1));
long armedAt = backoff.entry("g").nextAttemptMs();
// A second call inside the window does not arm and does not extend
the window.
@@ -119,34 +174,47 @@ public class StreamsGroupTopologyDescriptionBackoffTest {
@Test
public void testArmIfNotActiveReArmsAfterEpochAdvance() {
MockTime time = new MockTime();
- StreamsGroupTopologyDescriptionBackoff backoff = new
StreamsGroupTopologyDescriptionBackoff(time);
+ StreamsGroupTopologyDescriptionBackoff backoff =
deterministicBackoff(time);
assertTrue(backoff.armIfNotActive("g", 1));
// A query at the new epoch sees no active window and arms a fresh one.
assertTrue(backoff.armIfNotActive("g", 2));
assertEquals(2, backoff.entry("g").topologyEpoch());
- assertEquals(StreamsGroupTopologyDescriptionBackoff.INITIAL_DELAY_MS,
- backoff.entry("g").currentDelayMs());
+ assertEquals(0, backoff.entry("g").attempts());
}
@Test
- public void testArmIfNotActiveDoublesAfterExpiredWindowAtSameEpoch() {
+ public void
testArmIfNotActiveAdvancesAttemptsAfterExpiredWindowAtSameEpoch() {
// Heartbeats keep re-soliciting the same epoch because the client
never
// pushed (or the push was lost). The exponential chain must continue
across
- // those re-arms instead of resetting to INITIAL_DELAY_MS every cycle.
+ // those re-arms instead of resetting attempts every cycle.
MockTime time = new MockTime();
- StreamsGroupTopologyDescriptionBackoff backoff = new
StreamsGroupTopologyDescriptionBackoff(time);
+ StreamsGroupTopologyDescriptionBackoff backoff =
deterministicBackoff(time);
- long expected =
StreamsGroupTopologyDescriptionBackoff.INITIAL_DELAY_MS;
assertTrue(backoff.armIfNotActive("g", 1));
- assertEquals(expected, backoff.entry("g").currentDelayMs());
+ assertEquals(0, backoff.entry("g").attempts());
for (int i = 0; i < 20; i++) {
- time.sleep(backoff.entry("g").currentDelayMs());
+ time.sleep(backoff.entry("g").nextAttemptMs() -
time.milliseconds());
assertTrue(backoff.armIfNotActive("g", 1));
- expected = Math.min(expected * 2,
StreamsGroupTopologyDescriptionBackoff.MAX_DELAY_MS);
- assertEquals(expected, backoff.entry("g").currentDelayMs(),
"iteration " + i);
+ assertEquals(i + 1, backoff.entry("g").attempts(), "iteration " +
i);
}
- assertEquals(StreamsGroupTopologyDescriptionBackoff.MAX_DELAY_MS,
- backoff.entry("g").currentDelayMs());
+ }
+
+ @Test
+ public void testProductionWiringUsesJitter() {
+ // Sanity check that the no-arg constructor wires a jittered
ExponentialBackoff:
+ // the first delay should fall inside [(1 - JITTER), (1 + JITTER)] *
INITIAL_DELAY_MS,
+ // not be exactly INITIAL_DELAY_MS (with high probability — but we can
at least
+ // verify the entry exists and the upper-bound holds across many arms).
+ MockTime time = new MockTime();
+ StreamsGroupTopologyDescriptionBackoff backoff = new
StreamsGroupTopologyDescriptionBackoff(time);
+ backoff.armOrExtend("g", 1);
+ long delay = backoff.entry("g").nextAttemptMs() - time.milliseconds();
+ long lowerBound = (long) ((1 -
StreamsGroupTopologyDescriptionBackoff.JITTER)
+ * StreamsGroupTopologyDescriptionBackoff.INITIAL_DELAY_MS);
+ long upperBound = (long) ((1 +
StreamsGroupTopologyDescriptionBackoff.JITTER)
+ * StreamsGroupTopologyDescriptionBackoff.INITIAL_DELAY_MS);
+ assertTrue(delay >= lowerBound && delay <= upperBound,
+ "expected delay in [" + lowerBound + ", " + upperBound + "], got "
+ delay);
}
}
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionConverterTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionConverterTest.java
new file mode 100644
index 00000000000..811b7a425da
--- /dev/null
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionConverterTest.java
@@ -0,0 +1,197 @@
+/*
+ * 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.coordinator.group.streams;
+
+import org.apache.kafka.common.errors.InvalidRequestException;
+import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateRequestData;
+import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+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;
+
+public class StreamsGroupTopologyDescriptionConverterTest {
+
+ @Test
+ public void testConvertsAllThreeNodeKindsAndPreservesOrder() {
+
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode source
=
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode()
+ .setName("src")
+ .setNodeType((byte) 1)
+ .setSourceTopics(List.of("topic-a", "topic-b"))
+ .setSuccessors(List.of("proc-1", "proc-2"));
+
+
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode
processor =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode()
+ .setName("proc")
+ .setNodeType((byte) 2)
+ .setStores(List.of("store-x", "store-y"))
+ .setSuccessors(List.of("sink"));
+
+
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode sink =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode()
+ .setName("sink")
+ .setNodeType((byte) 3)
+ .setSinkTopic("out-topic")
+ .setSuccessors(List.of());
+
+
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionSubtopology
subtopology =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionSubtopology()
+ .setSubtopologyId("0")
+ .setNodes(List.of(source, processor, sink));
+
+ StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription
wire =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription()
+ .setSubtopologies(List.of(subtopology))
+ .setGlobalStores(List.of());
+
+ StreamsGroupTopologyDescription pojo =
StreamsGroupTopologyDescriptionConverter.fromRequest(wire);
+
+ assertEquals(1, pojo.subtopologies().size());
+ StreamsGroupTopologyDescription.Subtopology st =
pojo.subtopologies().iterator().next();
+ assertEquals("0", st.id());
+
+ Iterator<StreamsGroupTopologyDescription.Node> nodes =
st.nodes().iterator();
+ StreamsGroupTopologyDescription.Source src =
assertInstanceOf(StreamsGroupTopologyDescription.Source.class, nodes.next());
+ assertEquals("src", src.name());
+ assertEquals(List.of("topic-a", "topic-b"), new
ArrayList<>(src.topics()));
+ assertEquals(List.of("proc-1", "proc-2"), new
ArrayList<>(src.successors()));
+
+ StreamsGroupTopologyDescription.Processor proc =
assertInstanceOf(StreamsGroupTopologyDescription.Processor.class, nodes.next());
+ assertEquals("proc", proc.name());
+ assertEquals(List.of("store-x", "store-y"), new
ArrayList<>(proc.stores()));
+
+ StreamsGroupTopologyDescription.Sink sk =
assertInstanceOf(StreamsGroupTopologyDescription.Sink.class, nodes.next());
+ assertEquals("sink", sk.name());
+ assertTrue(sk.topic().isPresent());
+ assertEquals("out-topic", sk.topic().get());
+ }
+
+ @Test
+ public void testSinkWithoutTopicYieldsEmptyOptional() {
+
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode sink =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode()
+ .setName("sink")
+ .setNodeType((byte) 3)
+ .setSinkTopic(null)
+ .setSuccessors(List.of());
+
+ StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription
wire =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription()
+ .setSubtopologies(List.of(
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionSubtopology()
+ .setSubtopologyId("0")
+ .setNodes(List.of(sink))))
+ .setGlobalStores(List.of());
+
+ StreamsGroupTopologyDescription pojo =
StreamsGroupTopologyDescriptionConverter.fromRequest(wire);
+ StreamsGroupTopologyDescription.Sink sk =
(StreamsGroupTopologyDescription.Sink)
+ pojo.subtopologies().iterator().next().nodes().iterator().next();
+ assertTrue(sk.topic().isEmpty());
+ }
+
+ @Test
+ public void testGlobalStoreIsConverted() {
+
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode source
=
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode()
+ .setName("g-src")
+ .setNodeType((byte) 1)
+ .setSourceTopics(List.of("global-topic"))
+ .setSuccessors(List.of("g-proc"));
+
+
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode
processor =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode()
+ .setName("g-proc")
+ .setNodeType((byte) 2)
+ .setStores(List.of("global-store"))
+ .setSuccessors(List.of());
+
+
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionGlobalStore
gs =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionGlobalStore()
+ .setSource(source)
+ .setProcessor(processor);
+
+ StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription
wire =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription()
+ .setSubtopologies(List.of())
+ .setGlobalStores(List.of(gs));
+
+ StreamsGroupTopologyDescription pojo =
StreamsGroupTopologyDescriptionConverter.fromRequest(wire);
+ assertEquals(1, pojo.globalStores().size());
+ StreamsGroupTopologyDescription.GlobalStore converted =
pojo.globalStores().iterator().next();
+ assertEquals("g-src", converted.source().name());
+ assertEquals("g-proc", converted.processor().name());
+ }
+
+ @Test
+ public void testUnknownNodeTypeIsRejected() {
+
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode
unknown =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode()
+ .setName("?")
+ .setNodeType((byte) 99)
+ .setSuccessors(List.of());
+
+ StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription
wire =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription()
+ .setSubtopologies(List.of(
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionSubtopology()
+ .setSubtopologyId("0")
+ .setNodes(List.of(unknown))))
+ .setGlobalStores(List.of());
+
+ assertThrows(InvalidRequestException.class,
+ () -> StreamsGroupTopologyDescriptionConverter.fromRequest(wire));
+ }
+
+ @Test
+ public void testGlobalStoreWithMismatchedNodesIsRejected() {
+
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode
sourceA =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode()
+ .setName("a").setNodeType((byte) 1).setSuccessors(List.of());
+
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode
sourceB =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode()
+ .setName("b").setNodeType((byte) 1).setSuccessors(List.of());
+
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionGlobalStore
gs =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionGlobalStore()
+ .setSource(sourceA).setProcessor(sourceB);
+
+ StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription
wire =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription()
+ .setSubtopologies(List.of()).setGlobalStores(List.of(gs));
+
+ assertThrows(InvalidRequestException.class,
+ () -> StreamsGroupTopologyDescriptionConverter.fromRequest(wire));
+ }
+
+ @Test
+ public void testEmptyTopologyIsAccepted() {
+ StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription
wire =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription()
+ .setSubtopologies(List.of())
+ .setGlobalStores(List.of());
+ StreamsGroupTopologyDescription pojo =
StreamsGroupTopologyDescriptionConverter.fromRequest(wire);
+ assertTrue(pojo.subtopologies().isEmpty());
+ assertTrue(pojo.globalStores().isEmpty());
+ }
+}
