aliehsaeedii commented on code in PR #22777:
URL: https://github.com/apache/kafka/pull/22777#discussion_r3543004584
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -1866,17 +2040,40 @@ private CompletableFuture<Map<String, ApiError>>
deleteStreamsTopologyDescriptio
topicPartition,
(coordinator, lastCommittedOffset) ->
coordinator.streamsGroupsWithStoredTopologyDescription(groupIds,
lastCommittedOffset))
- .thenCompose(groupsWithStored ->
-
streamsGroupTopologyDescriptionManager.invokeDeleteTopologies(groupsWithStored)
- .thenApply(failures -> {
- // Clear back-off entries for every group whose plugin
state we
- // attempted to delete (regardless of plugin outcome):
the group is
- // about to be tombstoned on success and re-evaluated
by the next
- // heartbeat on failure, so any in-flight back-off
entry at the old
- // epoch is no longer load-bearing.
-
groupsWithStored.forEach(streamsGroupTopologyDescriptionManager::clearBackoffGroup);
- return failures;
- }))
+ .thenCompose(groupsWithStored -> {
+ // Common case: the batch holds no streams groups with stored
topology. Skip the
+ // mark entirely instead of scheduling a no-op write on the
shard's event loop.
+ if (groupsWithStored.isEmpty()) {
+ return CompletableFuture.<Map<String,
ApiError>>completedFuture(Map.of());
+ }
+ return markTopologyUncertainBatchAsync(topicPartition,
groupsWithStored)
+ .thenCompose(marked ->
+
streamsGroupTopologyDescriptionManager.invokeDeleteTopologies(marked)
+ .thenCompose(failures -> {
+ recordPluginDeleteOutcome(marked.size(),
failures.size());
+ Set<String> succeeded = new
LinkedHashSet<>(marked);
+ succeeded.removeAll(failures.keySet());
+ // Clear the push-path back-off only for
groups whose plugin delete
+ // succeeded (the group is about to be
tombstoned), mirroring the
+ // cleanup cycle. A failed delete leaves the
group at UNCERTAIN(-2),
+ // which re-solicits a push on the next
heartbeat — keep its back-off
+ // so a rejoining member does not immediately
hit the still-broken
+ // plugin, and let the interval-throttled
cleanup cycle retry it.
+
succeeded.forEach(streamsGroupTopologyDescriptionManager::clearBackoffGroup);
+ if (succeeded.isEmpty()) {
+ return
CompletableFuture.completedFuture(failures);
+ }
+ // Smart-finalize the successful deletes
before the tombstone,
Review Comment:
"a group revived between the mark and the plugin delete survives the
tombstone (NON_EMPTY_GROUP)" — only while the member is still present at
tombstone time. If the revive+push happens inside the delete window and the
member then leaves before the tombstone write executes, `validateDeleteGroup`
(emptiness-only, `StreamsGroup#validateDeleteGroup`) lets the tombstone
through: the smart-finalize's -2 repair record is erased together with the
group, and the plugin holds the raced push's data for a nonexistent group — an
unreclaimable orphan. The sweep path guards this with `shouldExpire` (stored !=
NONE defers) and the conversion path with the deferral check; explicit
DeleteGroups is the one tombstone path with no stored-epoch guard. Consider a
stored-epoch check in the streams `validateDeleteGroup` (successful deletes are
finalized to NONE before the tombstone, so the normal pipeline would pass it),
or document this as an accepted residual alongside the one in
`finalizeStoredDesc
riptionTopologyEpochAfterDelete`'s javadoc.
##########
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.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java:
##########
@@ -507,7 +530,8 @@ private CompletableFuture<Void> maybeAttachOne(
return null;
}
Integer storedEpoch =
result.storedDescriptionTopologyEpochs().get(describedGroup.groupId());
- if (storedEpoch == null || storedEpoch == -1 || storedEpoch !=
describedGroup.topology().epoch()) {
+ // <= NONE covers both NONE(-1) and UNCERTAIN(-2): nothing reliably
stored.
+ if (storedEpoch == null || storedEpoch <=
StreamsGroup.STORED_TOPOLOGY_EPOCH_NONE || storedEpoch !=
describedGroup.topology().epoch()) {
Review Comment:
Pre-existing, adjacent to this PR's scope (note, no action needed):
set-vs-set has no analog of the smart-finalize. Two pushes at epochs e1 < e2
can land in the plugin in reverse order; stored becomes `max` = e2 while the
plugin holds e1's description. `getTopology(groupId, e2)` then returns null
(the SPI is epoch-keyed), so describe reports NOT_STORED and nothing
re-solicits while stored == currentEpoch — the same NOT_STORED/no-re-solicit
liveness shape as the delete-vs-set race, but with no finalize analog to repair
it. The PR doesn't regress this — pre-PR wrote the epoch unconditionally — but
since the barrier/finalize pattern now exists for delete-vs-set, it may be
worth a follow-up JIRA for the set-vs-set ordering.
##########
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.
--
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]
