aliehsaeedii commented on code in PR #22777:
URL: https://github.com/apache/kafka/pull/22777#discussion_r3543004573
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -928,28 +968,43 @@ private void
recordPluginSetOutcome(StreamsGroupTopologyDescriptionManager.Plugi
}
/**
- * Batched conditional metadata write that clears {@code
StoredDescriptionTopologyEpoch}
- * for every entry in {@code expectedStoredEpochByGroupId}, but only for
the entries whose
- * persisted value still equals the supplied epoch. Mismatches and missing
groups are
- * silently ignored by the shard-side method. All groups in the batch must
hash to the same
- * __consumer_offsets partition (the caller guarantees this — the
eligibility scan is per
- * partition). Runtime write failures (NOT_COORDINATOR etc.) are logged
here and swallowed
- * so a single failed write does not poison the cycle's allOf — the next
cycle will retry
- * naturally because the persisted storedEpoch is still non-default.
+ * Batched UNCERTAIN(-2) barrier write before the cleanup cycle's plugin
delete. The shard-side
+ * method re-checks the latest state per group and returns only the subset
that is still a
+ * streams group and is now UNCERTAIN; revived or converted candidates
drop out so they are not
+ * deleted. All groups in the batch hash to the same __consumer_offsets
partition (the caller
+ * guarantees this — the eligibility scan is per partition).
*/
- private CompletableFuture<Void>
clearStoredDescriptionTopologyEpochBatchAsync(
- Map<String, Integer> expectedStoredEpochByGroupId
+ private CompletableFuture<Set<String>> markTopologyUncertainBatchAsync(
+ TopicPartition tp,
+ Set<String> groupIds
) {
- if (expectedStoredEpochByGroupId.isEmpty()) return
CompletableFuture.completedFuture(null);
- TopicPartition tp =
topicPartitionFor(expectedStoredEpochByGroupId.keySet().iterator().next());
return runtime.scheduleWriteOperation(
- "clear-stored-topology-epoch",
+ "mark-topology-uncertain-batch",
+ tp,
+ coordinator ->
coordinator.markStoredDescriptionTopologyEpochUncertainBatch(groupIds));
+ }
+
+ /**
+ * Batched smart-finalize write after the cleanup cycle's plugin delete.
Per group: if stored
+ * is still UNCERTAIN no push raced and it is cleared to NONE (the next
sweep tombstones it); if
+ * stored advanced past UNCERTAIN a push raced our delete and it is forced
back to UNCERTAIN to
+ * re-solicit. All groups in the batch hash to the same __consumer_offsets
partition. Runtime
+ * write failures (NOT_COORDINATOR etc.) are logged here and swallowed so
a single failed write
+ * does not poison the cycle's allOf — the next cycle retries because the
persisted storedEpoch
+ * is still non-default.
+ */
+ private CompletableFuture<Void> finalizeAfterDeleteBatchAsync(
+ TopicPartition tp,
+ Set<String> groupIds
+ ) {
+ return runtime.<Void>scheduleWriteOperation(
+ "finalize-stored-topology-epoch-after-delete-batch",
tp,
- coordinator ->
coordinator.clearStoredDescriptionTopologyEpochBatch(expectedStoredEpochByGroupId)
+ coordinator ->
coordinator.finalizeStoredDescriptionTopologyEpochAfterDeleteBatch(groupIds)
).handle((__, throwable) -> {
if (throwable != null) {
- log.warn("Failed to clear StoredDescriptionTopologyEpoch for
groups {} on partition {}; the next cleanup cycle will retry.",
- expectedStoredEpochByGroupId.keySet(), tp, throwable);
+ log.warn("Failed to finalize StoredDescriptionTopologyEpoch
for groups {} on partition {}; "
Review Comment:
The smart-finalize is the *repair* step of the barrier protocol, but it is
one-shot and its failure is swallowed here. The warn's claim ("the next cleanup
cycle will retry") only holds while the group stays empty. Interleaving with no
repair path: cleanup marks -2 and starts `plugin.deleteTopology`; the group is
revived and a push runs — its `setTopology` lands first, then our
still-in-flight delete lands after it and empties the plugin, while the push's
epoch write commits `max(-2, e') = e'`; then this finalize write fails (e.g.
NOT_COORDINATOR during a shard move). Result: stored = e' over an emptied
plugin, group non-empty — the cleanup scan skips it, solicitation is suppressed
(`storedEpoch == currentEpoch`), and describe reports NOT_STORED until the next
topology-epoch bump. That contradicts the PR's "any failure after the barrier
leaves the group uncertain and therefore self-healing" claim: the barrier
protects the step *before* the plugin op, but the repair *after* it h
as no durability/retry. Worth either documenting as an accepted residual (and
rewording the warn), or making the finalize retryable. Also: no test stubs
`finalize-stored-topology-epoch-after-delete-batch` with a failed future — the
mark-write twin has one (`testCleanupCycleSwallowsMarkWriteFailure`), the
finalize swallow does not.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -1268,6 +1333,115 @@ public CompletableFuture<JoinGroupResponseData>
joinGroup(
return responseFuture;
}
+ /**
+ * Converting an empty streams group to classic would orphan the plugin's
topology, so the
+ * join detected cleanup is needed: delete the topology (behind a durable
UNCERTAIN(-2)
+ * barrier) and re-run the join, which then converts because cleanup has
been handled.
+ *
+ * <p>Throttle first: on {@code REBALANCE_IN_PROGRESS} the classic client
retries the join
+ * immediately ({@code RebalanceInProgressException} skips its retry
back-off), so a broken
+ * plugin would otherwise be hit with {@code deleteTopology} in a tight
loop. While the window
+ * armed by a previous failed conversion delete is in effect, fail fast
without touching the
+ * plugin or scheduling the mark; the interval-throttled cleanup cycle
reclaims the group in
+ * the meantime.
+ */
+ private CompletableFuture<Void> cleanupTopologyBeforeConversion(
+ AuthorizableRequestContext context,
+ JoinGroupRequestData request,
+ CompletableFuture<JoinGroupResponseData> responseFuture,
+ TopicPartition tp
+ ) {
+ if
(streamsGroupTopologyDescriptionManager.isConversionDeleteThrottled(request.groupId()))
{
+ failJoinRetriably(request, responseFuture);
+ return CompletableFuture.completedFuture(null);
+ }
+ return markTopologyUncertainAsync(tp, request.groupId(), false)
+ .thenCompose(marked -> {
+ if (!marked) {
+ // The group changed underneath us (revived, converted, or
removed) between
+ // the join's cleanup check and the barrier write: no
barrier exists, so
+ // running the plugin delete could wipe a live group's
topology. Fail the
+ // join with a retriable error and let the client retry
against the latest
+ // group state.
+ failJoinRetriably(request, responseFuture);
+ return CompletableFuture.<Void>completedFuture(null);
+ }
+ return
streamsGroupTopologyDescriptionManager.invokeDeleteTopologies(Set.of(request.groupId()))
+ .thenCompose(failures -> {
+ recordPluginDeleteOutcome(1, failures.size());
+ if (!failures.isEmpty()) {
+ // Plugin delete failed: leave the group a streams
group at UNCERTAIN(-2)
+ // (reclaimable by the cleanup cycle and
re-soliciting), arm the
+ // conversion-delete throttle so the client's
immediate join retries do
+ // not hammer the broken plugin, and fail the join
with a retriable
+ // error instead of converting over orphaned
plugin data.
+
streamsGroupTopologyDescriptionManager.throttleConversionDelete(request.groupId());
+ failJoinRetriably(request, responseFuture);
+ return
CompletableFuture.<Void>completedFuture(null);
+ }
+
streamsGroupTopologyDescriptionManager.clearBackoffGroup(request.groupId());
+ // Smart-finalize after the re-join: a no-op once the
group has been
+ // converted (it is no longer a streams group). It
only writes for a group
+ // revived between the barrier and the re-join — the
re-join then rejects
+ // with INCONSISTENT_GROUP_PROTOCOL and, without the
finalize, a raced
+ // push's epoch write would land on stored ==
UNCERTAIN and record a real
+ // epoch over the plugin this delete just emptied.
+ return runClassicGroupJoin(context, request,
responseFuture, tp, true)
Review Comment:
The re-join trusts `topologyCleanupHandled=true` even when the stored epoch
has moved past UNCERTAIN, which re-opens the leak this PR closes elsewhere.
Window: after our `deleteTopology` succeeds, a streams member revives the
group, pushes (its mark no-ops at -2; its `setTopology` lands after our delete;
its epoch write commits stored = e'), and leaves — the group is empty again by
the time this re-join executes. The flag suppresses the deferral check in
`classicGroupJoin`, so `maybeDeleteEmptyStreamsGroup` tombstones the streams
metadata over stored = e' while the plugin still holds e''s data; the finalize
afterwards no-ops because the group is already classic. Nothing reclaims the
plugin entry (the cleanup scan iterates live streams groups only). The comment
above covers only the still-revived case (INCONSISTENT_GROUP_PROTOCOL), not the
emptied-again case. A cheap close: with `topologyCleanupHandled=true`, still
re-signal cleanup when `stored >= 0` — a raced push is the only
way to reach a real epoch behind the barrier — and trust the flag only for
stored == -2.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
