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/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.
--
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]
