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 e835bc4e3ea KAFKA-20623: Follow-up cleanups for the streams group
topology description plugin (#22616)
e835bc4e3ea is described below
commit e835bc4e3eadde40d5423bccae0054dbb199f4a8
Author: Lucas Brutschy <[email protected]>
AuthorDate: Thu Jun 18 16:58:26 2026 +0200
KAFKA-20623: Follow-up cleanups for the streams group topology description
plugin (#22616)
Follow-up cleanups for the streams group topology description plugin
(KIP-1331), addressing review feedback from the 2/3 and 3/3 PRs (#22552,
#22554) that was deferred at merge time.
The DeleteGroups pre-delete path is hardened. A failure of the streams
topology pre-delete read now goes through handleOperationException, so
it reports the same translated, retriable error codes as the rest of the
delete pipeline rather than a raw code, and the plugin's own failure
message is no longer forwarded to the client. The stored and failed
topology epochs are only ever advanced, never regressed, which closes a
window where a slow push committing after the group's epoch advanced —
or after a concurrent higher-epoch push — could move them backwards.
The push RPC handler is simplified. The back-off disposition is now
acted on where it is known rather than threaded through the future chain
as a value object, which removes the SetTopologyOutcome, BackoffAction,
applySetTopologyBackoff and outcomeForPostPluginWrite machinery; the
post-write back-off handling lives on
StreamsGroupTopologyDescriptionManager next to the back-off it mutates,
while the runtime chain stays in GroupCoordinatorService. Suppressing
solicitation for a departing member moves out of GroupMetadataManager,
which previously built a sentinel result, into the service-layer gate
that already owns the solicitation policy, so leave and fence results
carry the group's real epochs like any other heartbeat.
Smaller cleanups round it out: the per-group back-off uses
ConcurrentHashMap.compute consistently rather than a hand-rolled
compare-and-swap loop, the pre-delete group filter uses a non-throwing
lookup instead of catching an exception per non-streams group, and the
boolean-flag epoch-write method is split into named methods.
One known limitation is intentionally left for a separate follow-up: the
broker-local back-off map is only evicted on DeleteGroups, so groups
removed by session expiry, partition unload, or tombstone-via-replay
retain their entry until the group id is reused.
Reviewers: TengYao Chi <[email protected]>
---
.../coordinator/group/GroupCoordinatorService.java | 170 +++++++--------------
.../coordinator/group/GroupCoordinatorShard.java | 31 ++--
.../coordinator/group/GroupMetadataManager.java | 91 +++++++----
.../group/streams/StreamsGroupHeartbeatResult.java | 25 +--
.../StreamsGroupTopologyDescriptionBackoff.java | 22 ++-
.../StreamsGroupTopologyDescriptionManager.java | 68 +++++++--
...pCoordinatorServiceTopologyDescriptionTest.java | 28 +++-
7 files changed, 231 insertions(+), 204 deletions(-)
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 2708b7b261b..80a3f29ff61 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,9 +20,6 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.compress.Compression;
import org.apache.kafka.common.config.TopicConfig;
-import org.apache.kafka.common.errors.CoordinatorLoadInProgressException;
-import org.apache.kafka.common.errors.CoordinatorNotAvailableException;
-import org.apache.kafka.common.errors.GroupIdNotFoundException;
import org.apache.kafka.common.errors.InvalidRequestException;
import org.apache.kafka.common.errors.NotCoordinatorException;
import org.apache.kafka.common.errors.StreamsInvalidTopologyException;
@@ -633,7 +630,7 @@ public class GroupCoordinatorService implements
GroupCoordinator {
) {
if (!isActive.get()) {
return CompletableFuture.completedFuture(
- StreamsGroupHeartbeatResult.withoutEpochContext(
+ StreamsGroupHeartbeatResult.forError(
new
StreamsGroupHeartbeatResponseData().setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())
)
);
@@ -645,7 +642,7 @@ public class GroupCoordinatorService implements
GroupCoordinator {
} catch (Throwable ex) {
ApiError apiError = ApiError.fromThrowable(ex);
return CompletableFuture.completedFuture(
- StreamsGroupHeartbeatResult.withoutEpochContext(
+ StreamsGroupHeartbeatResult.forError(
new StreamsGroupHeartbeatResponseData()
.setErrorCode(apiError.error().code())
.setErrorMessage(apiError.message())
@@ -662,7 +659,7 @@ public class GroupCoordinatorService implements
GroupCoordinator {
heartbeat = heartbeat.thenApply(result -> {
try {
return
streamsGroupTopologyDescriptionManager.maybeSetTopologyDescriptionRequired(
- result, request.groupId(), context.requestVersion());
+ result, request.groupId(), context.requestVersion(),
request.memberEpoch());
} 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
@@ -682,7 +679,7 @@ public class GroupCoordinatorService implements
GroupCoordinator {
request,
exception,
(error, message) ->
- StreamsGroupHeartbeatResult.withoutEpochContext(
+ StreamsGroupHeartbeatResult.forError(
new StreamsGroupHeartbeatResponseData()
.setErrorCode(error.code())
.setErrorMessage(message)
@@ -725,14 +722,12 @@ public class GroupCoordinatorService implements
GroupCoordinator {
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.
+ // The back-off is mutated where the disposition is known: pre-plugin
failures (validate /
+ // convert / runtime) never reach the arming code, so a fenced or
unauthorized caller
+ // cannot grief the back-off; a transient plugin failure arms it; and
the post-plugin
+ // bookkeeping write clears it on success, drops the whole entry if
the group was deleted
+ // underneath us, leaves it alone on a coordinator-moved error, or
arms it (see
+ // StreamsGroupTopologyDescriptionManager#completeEpochWrite).
return runtime.scheduleReadOperation(
"streams-group-topology-description-validate",
tp,
@@ -744,15 +739,30 @@ public class GroupCoordinatorService implements
GroupCoordinator {
.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(pluginOutcome -> switch (pluginOutcome.kind()) {
+ case SUCCESS -> runtime.scheduleWriteOperation(
+ "streams-group-set-stored-topology-epoch",
+ tp,
+ coordinator ->
coordinator.setStoredDescriptionTopologyEpoch(groupId, pushedEpoch)
+ ).handle((unused, throwable) ->
streamsGroupTopologyDescriptionManager.completeEpochWrite(
+ groupId, pushedEpoch, throwable,
+ new StreamsGroupTopologyDescriptionUpdateResponseData()));
+ case PERMANENT -> runtime.scheduleWriteOperation(
+ "streams-group-set-failed-topology-epoch",
+ tp,
+ coordinator ->
coordinator.setFailedDescriptionTopologyEpoch(groupId, pushedEpoch)
+ ).handle((unused, throwable) ->
streamsGroupTopologyDescriptionManager.completeEpochWrite(
+ groupId, pushedEpoch, throwable,
+ new StreamsGroupTopologyDescriptionUpdateResponseData()
+
.setErrorCode(Errors.STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED.code())
+ .setErrorMessage(pluginOutcome.message())));
+ case TRANSIENT -> {
+ streamsGroupTopologyDescriptionManager.armBackoff(groupId,
pushedEpoch);
+ yield CompletableFuture.completedFuture(new
StreamsGroupTopologyDescriptionUpdateResponseData()
+
.setErrorCode(Errors.STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED.code())
+ .setErrorMessage(pluginOutcome.message()));
+ }
})
- .thenCompose(outcome -> outcome.failure() == null
- ? CompletableFuture.completedFuture(outcome.response())
- : CompletableFuture.failedFuture(outcome.failure()))
.exceptionally(exception -> handleOperationException(
"streams-group-topology-description-update",
request,
@@ -764,86 +774,6 @@ public class GroupCoordinatorService implements
GroupCoordinator {
));
}
- 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);
- }
- Throwable cause = Errors.maybeUnwrapException(throwable);
- // The group was deleted between the plugin call and the bookkeeping
write — the
- // push already took effect at the plugin and no live group remains to
throttle,
- // so drop the orphaned back-off entry instead of arming one nobody
will clear.
- if (cause instanceof GroupIdNotFoundException) {
- return new SetTopologyOutcome(null, BackoffAction.CLEAR_GROUP,
throwable);
- }
- // This broker stopped being the coordinator between the plugin call
and the
- // bookkeeping write. The client will retry against the new
coordinator, which
- // holds no back-off entry of its own; arming a broker-wide entry here
would
- // leak until expiry and could suppress a legitimate solicitation if
the group
- // ever migrates back. The retry that lands on the new coordinator is
what
- // re-establishes convergence, not a back-off window on the broker
that bowed out.
- if (cause instanceof NotCoordinatorException
- || cause instanceof CoordinatorLoadInProgressException
- || cause instanceof CoordinatorNotAvailableException) {
- return new SetTopologyOutcome(null, BackoffAction.NOOP, throwable);
- }
- return new SetTopologyOutcome(null, BackoffAction.ARM, throwable);
- }
-
- 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 {
@@ -1641,7 +1571,7 @@ public class GroupCoordinatorService implements
GroupCoordinator {
return
CompletableFuture.completedFuture(deletableGroupResults);
}
- return runPrePluginDelete(topicPartition, retainedGroupIds)
+ return deleteStreamsTopologyDescriptions(topicPartition,
retainedGroupIds)
.thenCompose(streamsErrMap -> {
List<String> afterStreams =
filterStreamsTopologyErrors(
streamsErrMap, retainedGroupIds,
deletableGroupResults);
@@ -1654,11 +1584,11 @@ public class GroupCoordinatorService implements
GroupCoordinator {
})
.exceptionally(exception -> {
// Defensive net for any uncaught synchronous throw in
the
- // post-runPrePluginDelete stage. Without this, the
exception would
+ // post-deleteStreamsTopologyDescriptions stage.
Without this, the exception would
// propagate through FutureUtils.combineFutures.join()
and fail the
// whole cross-partition DeleteGroups response —
including groups on
// other partitions that already succeeded. Runtime
read failures
- // inside runPrePluginDelete are absorbed there, so
they never reach
+ // inside deleteStreamsTopologyDescriptions are
absorbed there, so they never reach
// this branch; what we are catching here is the
synchronous stages
// (filterStreamsTopologyErrors etc.). Fold the
exception into
// per-group failures for any retainedGroupIds not yet
recorded.
@@ -1752,7 +1682,7 @@ public class GroupCoordinatorService implements
GroupCoordinator {
* pass through the more specific runtime error rather than collapsing
everything to
* {@code GROUP_DELETION_FAILED}.
*/
- private CompletableFuture<Map<String, ApiError>> runPrePluginDelete(
+ private CompletableFuture<Map<String, ApiError>>
deleteStreamsTopologyDescriptions(
TopicPartition topicPartition,
List<String> groupIds
) {
@@ -1775,17 +1705,21 @@ public class GroupCoordinatorService implements
GroupCoordinator {
groupsWithStored.forEach(streamsGroupTopologyDescriptionManager::clearBackoffGroup);
return failures;
}))
- .exceptionally(exception -> {
- // Use ApiError.fromThrowable so the error reported to the
client matches what
- // the sibling DeleteGroups paths return: it unwraps
CompletionException so the
- // ErrorMessage is the cause's own message rather than the
FQCN-prefixed
- // toString(), and it suppresses the message for
UNKNOWN_SERVER_ERROR so we do
- // not leak plugin internals.
- ApiError apiError = ApiError.fromThrowable(exception);
- Map<String, ApiError> failures = new HashMap<>();
- groupIds.forEach(id -> failures.put(id, apiError));
- return failures;
- });
+ .exceptionally(exception -> handleOperationException(
+ // Translate coordinator errors so a read failure reports the
same retriable code
+ // as the rest of the DeleteGroups pipeline (e.g.
NOT_LEADER_OR_FOLLOWER ->
+ // NOT_COORDINATOR), and unwrap/sanitize the message.
+ "streams-group-topology-pre-delete",
+ groupIds,
+ exception,
+ (error, message) -> {
+ ApiError apiError = new ApiError(error, message);
+ Map<String, ApiError> failures = new HashMap<>();
+ groupIds.forEach(id -> failures.put(id, apiError));
+ return failures;
+ },
+ log
+ ));
}
/**
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 d449f4b18f2..b3dc8efa6c4 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
@@ -941,22 +941,33 @@ public class GroupCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
}
/**
- * 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).
+ * Advance {@code StoredDescriptionTopologyEpoch} after a successful
plugin {@code setTopology}.
*
- * @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.
+ * @param groupId The streams group id.
+ * @param pushedEpoch The topology epoch on the push that just completed.
* @return A coordinator result carrying the metadata record.
* @throws GroupIdNotFoundException if the streams group no longer exists.
*/
- public CoordinatorResult<Void, CoordinatorRecord>
streamsGroupSetTopologyDescriptionEpoch(
+ public CoordinatorResult<Void, CoordinatorRecord>
setStoredDescriptionTopologyEpoch(
String groupId,
- int pushedEpoch,
- boolean permanentFailure
+ int pushedEpoch
+ ) {
+ return groupMetadataManager.setStoredDescriptionTopologyEpoch(groupId,
pushedEpoch);
+ }
+
+ /**
+ * Advance {@code FailedDescriptionTopologyEpoch} after a permanent plugin
failure.
+ *
+ * @param groupId The streams group id.
+ * @param pushedEpoch The topology epoch on the push that just completed.
+ * @return A coordinator result carrying the metadata record.
+ * @throws GroupIdNotFoundException if the streams group no longer exists.
+ */
+ public CoordinatorResult<Void, CoordinatorRecord>
setFailedDescriptionTopologyEpoch(
+ String groupId,
+ int pushedEpoch
) {
- return
groupMetadataManager.streamsGroupSetTopologyDescriptionEpoch(groupId,
pushedEpoch, permanentFailure);
+ return groupMetadataManager.setFailedDescriptionTopologyEpoch(groupId,
pushedEpoch);
}
/**
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 5a32fa8f1cb..ef62f63eca1 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
@@ -4470,15 +4470,16 @@ 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,
StreamsGroupHeartbeatResult.withoutEpochContext(response));
+ return streamsGroupFenceMember(group, member, new
StreamsGroupHeartbeatResult(
+ response,
+ Map.of(),
+ group.currentTopologyEpoch(),
+ group.storedDescriptionTopologyEpoch(),
+ group.failedDescriptionTopologyEpoch()
+ ));
} else {
StreamsGroupMember member = group.staticMember(instanceId);
throwIfStaticMemberIsUnknown(member, instanceId);
@@ -4490,7 +4491,13 @@ 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,
StreamsGroupHeartbeatResult.withoutEpochContext(response));
+ return streamsGroupFenceMember(group, member, new
StreamsGroupHeartbeatResult(
+ response,
+ Map.of(),
+ group.currentTopologyEpoch(),
+ group.storedDescriptionTopologyEpoch(),
+ group.failedDescriptionTopologyEpoch()
+ ));
}
}
}
@@ -4557,12 +4564,15 @@ 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),
- StreamsGroupHeartbeatResult.withoutEpochContext(response)
+ new StreamsGroupHeartbeatResult(
+ response,
+ Map.of(),
+ group.currentTopologyEpoch(),
+ group.storedDescriptionTopologyEpoch(),
+ group.failedDescriptionTopologyEpoch()
+ )
);
}
@@ -8408,41 +8418,64 @@ public class GroupMetadataManager {
) {
Set<String> withStored = new HashSet<>();
for (String groupId : groupIds) {
- try {
- StreamsGroup group = streamsGroup(groupId, committedOffset);
- if (group.storedDescriptionTopologyEpoch(committedOffset) !=
-1) {
- withStored.add(groupId);
- }
- } catch (GroupIdNotFoundException ignored) {
- // Not a streams group on this shard; nothing to clean up.
+ // Non-throwing lookup + type check: silently skip absent or
non-streams groups.
+ Group group = groups.get(groupId, committedOffset);
+ if (group != null
+ && group.type() == STREAMS
+ && ((StreamsGroup)
group).storedDescriptionTopologyEpoch(committedOffset) != -1) {
+ withStored.add(groupId);
}
}
return withStored;
}
/**
- * Persist the outcome of a topology description plugin call for a streams
group.
+ * Advance {@code StoredDescriptionTopologyEpoch} to {@code pushedEpoch}
after a successful
+ * plugin {@code setTopology}, so subsequent heartbeats at the same epoch
do not re-solicit.
*
- * <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.
+ * @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>
setStoredDescriptionTopologyEpoch(
+ String groupId,
+ int pushedEpoch
+ ) throws GroupIdNotFoundException {
+ return updateTopologyDescriptionEpochs(groupId, pushedEpoch, false);
+ }
+
+ /**
+ * Advance {@code FailedDescriptionTopologyEpoch} to {@code pushedEpoch}
after a permanent
+ * plugin failure, 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.
+ * @param groupId The streams group id.
+ * @param pushedEpoch The topology epoch on the push that just completed.
* @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(
+ public CoordinatorResult<Void, CoordinatorRecord>
setFailedDescriptionTopologyEpoch(
+ String groupId,
+ int pushedEpoch
+ ) throws GroupIdNotFoundException {
+ return updateTopologyDescriptionEpochs(groupId, pushedEpoch, true);
+ }
+
+ private CoordinatorResult<Void, CoordinatorRecord>
updateTopologyDescriptionEpochs(
String groupId,
int pushedEpoch,
boolean permanentFailure
) throws GroupIdNotFoundException {
StreamsGroup group = streamsGroup(groupId);
- int newStored = permanentFailure ?
group.storedDescriptionTopologyEpoch() : pushedEpoch;
- int newFailed = permanentFailure ? pushedEpoch :
group.failedDescriptionTopologyEpoch();
+ // Only advance these epochs, never regress them: a stale push
committing after the group
+ // advanced (or after a concurrent higher-epoch push) must not move
stored/failed back.
+ int newStored = permanentFailure
+ ? group.storedDescriptionTopologyEpoch()
+ : Math.max(group.storedDescriptionTopologyEpoch(), pushedEpoch);
+ int newFailed = permanentFailure
+ ? Math.max(group.failedDescriptionTopologyEpoch(), pushedEpoch)
+ : group.failedDescriptionTopologyEpoch();
CoordinatorRecord record = newStreamsGroupMetadataRecord(
groupId,
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 28fc1e844f2..0a2638e753f 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,25 +55,14 @@ public record StreamsGroupHeartbeatResult(
}
/**
- * 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.
+ * Build a heartbeat result for an error response — a failure-fast path in
+ * {@code GroupCoordinatorService#streamsGroupHeartbeat} (broker not
active, request
+ * validation rejected, or a runtime error translated by {@code
handleOperationException}).
+ * No group is resolved, so there are no internal topics to create and no
epoch context:
+ * all three epoch fields are -1, and {@code
maybeSetTopologyDescriptionRequired}
+ * short-circuits on the error code anyway. Callers must pass a response
carrying an error.
*/
- public static StreamsGroupHeartbeatResult
withoutEpochContext(StreamsGroupHeartbeatResponseData data) {
+ public static StreamsGroupHeartbeatResult
forError(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 2d4035aea95..3bf2c3f696f 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
@@ -20,6 +20,7 @@ import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.internals.ExponentialBackoff;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
/**
* In-memory per-group back-off that throttles broker re-solicitation of a
topology
@@ -72,23 +73,18 @@ public class StreamsGroupTopologyDescriptionBackoff {
* or the push was lost in flight before {@link #armOrExtend} could run).
*/
public boolean armIfNotActive(String groupId, int topologyEpoch) {
- while (true) {
- final long now = time.milliseconds();
- Entry existing = state.get(groupId);
+ final long now = time.milliseconds();
+ final AtomicBoolean armed = new AtomicBoolean(false);
+ state.compute(groupId, (key, existing) -> {
if (existing != null
&& existing.topologyEpoch() == topologyEpoch
&& now < existing.nextAttemptMs()) {
- return false;
- }
- 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 existing;
}
- // Lost a race with a concurrent mutation; retry with the fresh
state.
- }
+ armed.set(true);
+ return computeNextEntry(existing, topologyEpoch, now);
+ });
+ return armed.get();
}
/**
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 9b52a1506d3..93cf81504e3 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
@@ -16,7 +16,12 @@
*/
package org.apache.kafka.coordinator.group.streams;
+import org.apache.kafka.common.errors.CoordinatorLoadInProgressException;
+import org.apache.kafka.common.errors.CoordinatorNotAvailableException;
+import org.apache.kafka.common.errors.GroupIdNotFoundException;
+import org.apache.kafka.common.errors.NotCoordinatorException;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
+import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateResponseData;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.ApiError;
import org.apache.kafka.common.requests.StreamsGroupHeartbeatResponse.Status;
@@ -33,6 +38,7 @@ import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
/**
* Broker-level component that owns the streams-group topology description
plugin
@@ -102,9 +108,13 @@ public class StreamsGroupTopologyDescriptionManager
implements AutoCloseable {
public StreamsGroupHeartbeatResult maybeSetTopologyDescriptionRequired(
StreamsGroupHeartbeatResult result,
String groupId,
- int apiVersion
+ int apiVersion,
+ int memberEpoch
) {
- if (apiVersion < 1 || plugin.isEmpty()) {
+ // Do not solicit a push from a departing member (a leave heartbeat
carries a negative
+ // member epoch): arming the back-off on its behalf would only delay
solicitation for the
+ // rest of the group.
+ if (apiVersion < 1 || plugin.isEmpty() || memberEpoch < 0) {
return result;
}
StreamsGroupHeartbeatResponseData response = result.data();
@@ -185,19 +195,47 @@ public class StreamsGroupTopologyDescriptionManager
implements AutoCloseable {
}
/**
- * 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}.
+ * Settle the per-group back-off after the bookkeeping write that records
a push outcome
+ * (the stored epoch on success, the failed epoch on a permanent failure)
completes, and
+ * return the response to send to the client — or rethrow the write
failure so the service's
+ * terminal handler maps it.
+ *
+ * <p>On a clean write the back-off is cleared for this epoch
(epoch-scoped so a late callback
+ * at an old epoch cannot wipe a window a concurrent heartbeat armed at
the advanced epoch);
+ * if the group was deleted underneath the write its whole entry is
dropped; a coordinator-moved
+ * error leaves the back-off untouched (the new coordinator owns
convergence once the client
+ * retries); any other failure arms it so the next heartbeat re-solicits.
*/
- public void clearBackoff(String groupId, int topologyEpoch) {
- backoff.clear(groupId, topologyEpoch);
+ public StreamsGroupTopologyDescriptionUpdateResponseData
completeEpochWrite(
+ String groupId,
+ int topologyEpoch,
+ Throwable writeException,
+ StreamsGroupTopologyDescriptionUpdateResponseData responseOnCommit
+ ) {
+ if (writeException == null) {
+ backoff.clear(groupId, topologyEpoch);
+ return responseOnCommit;
+ }
+ Throwable cause = Errors.maybeUnwrapException(writeException);
+ if (cause instanceof GroupIdNotFoundException) {
+ backoff.clearGroup(groupId);
+ } else if (cause instanceof NotCoordinatorException
+ || cause instanceof CoordinatorLoadInProgressException
+ || cause instanceof CoordinatorNotAvailableException) {
+ // Coordinator moved between the plugin call and the write; the
new coordinator owns
+ // convergence after the client retries, so leave the back-off
alone.
+ } else {
+ backoff.armOrExtend(groupId, topologyEpoch);
+ }
+ throw new CompletionException(writeException);
}
/**
- * 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
+ * Drop the back-off entry for a group unconditionally. Currently called
when a group is
+ * removed via explicit DeleteGroups and on a post-plugin write failing
with
+ * GroupIdNotFoundException. NOTE: groups removed by other lifecycle paths
(session expiry,
+ * partition unload, tombstone-via-replay) are not yet wired to this, so
their back-off
+ * entries can leak until the group id is reused. Delegates to
* {@link StreamsGroupTopologyDescriptionBackoff#clearGroup}.
*/
public void clearBackoffGroup(String groupId) {
@@ -254,9 +292,11 @@ public class StreamsGroupTopologyDescriptionManager
implements AutoCloseable {
if (throwable == null) {
return null;
}
- Throwable cause = Errors.maybeUnwrapException(throwable);
- String message = cause != null ? cause.getMessage() : "Plugin failure
(no cause).";
- return Map.entry(groupId, new ApiError(Errors.GROUP_DELETION_FAILED,
message));
+ // Do not forward the plugin's raw exception message to the client: it
can be null and
+ // may leak plugin internals (the ErrorMessage is serialized at
DeleteGroups v3+). Use a
+ // fixed generic message, mirroring invokeSetTopology.
+ return Map.entry(groupId, new ApiError(Errors.GROUP_DELETION_FAILED,
+ "Topology description plugin failed to delete the topology."));
}
// Visible for testing.
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 20ac95683b6..8fe7b89a60b 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
@@ -495,6 +495,29 @@ public class
GroupCoordinatorServiceTopologyDescriptionTest {
assertTrue(result.data().topologyDescriptionRequired());
}
+ @Test
+ public void testHeartbeatSkipsFlagForDepartingMember() throws Exception {
+ // A leave heartbeat carries a negative member epoch. Even though the
result shows the
+ // stored epoch lagging (which would otherwise solicit a push), the
gate must not arm
+ // the back-off or set the flag for a member on its way out.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(runtime.scheduleWriteOperation(
+ eq("streams-group-heartbeat"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(
+ new StreamsGroupHeartbeatResult(new
StreamsGroupHeartbeatResponseData(), Map.of(), 5, -1, -1)));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ StreamsGroupHeartbeatResult result = service.streamsGroupHeartbeat(
+ requestContext(ApiKeys.STREAMS_GROUP_HEARTBEAT),
+ validHeartbeatRequest().setMemberEpoch(-1)
+ ).get(5, TimeUnit.SECONDS);
+
+ assertFalse(result.data().topologyDescriptionRequired());
+ }
+
@Test
public void testHeartbeatArmSuppressReSolicitCycle() throws Exception {
// End-to-end exercise of the arm → suppress → re-solicit cycle
through the
@@ -710,7 +733,8 @@ public class GroupCoordinatorServiceTopologyDescriptionTest
{
DeleteGroupsResponseData.DeletableGroupResult result =
results.find("foo");
assertNotNull(result);
assertEquals(Errors.GROUP_DELETION_FAILED.code(), result.errorCode());
- assertEquals("plugin offline", result.errorMessage());
+ // The raw plugin message ("plugin offline") must not be forwarded to
the client.
+ assertEquals("Topology description plugin failed to delete the
topology.", result.errorMessage());
verify(runtime, never()).scheduleWriteOperation(
eq("delete-groups"), any(), any());
}
@@ -886,7 +910,7 @@ public class GroupCoordinatorServiceTopologyDescriptionTest
{
DeleteGroupsResponseData.DeletableGroupResult badResult =
results.find("bad");
assertNotNull(badResult);
assertEquals(Errors.GROUP_DELETION_FAILED.code(),
badResult.errorCode());
- assertEquals("rejected", badResult.errorMessage());
+ assertEquals("Topology description plugin failed to delete the
topology.", badResult.errorMessage());
}
private static StreamsGroupTopologyDescriptionUpdateRequestData
validUpdateRequest() {
