Copilot commented on code in PR #23566:
URL: https://github.com/apache/kafka/pull/23566#discussion_r4087960707
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java:
##########
@@ -223,6 +304,103 @@ public boolean isPluginConfigured() {
return plugin.isPresent();
}
+ /**
+ * One topology-description cleanup cycle: (1) reads the streams groups
eligible for
+ * plugin-side cleanup, one batch per locality {@link
TopologyDescriptionRuntime} splits the
+ * scan across; (2) writes a durable UNCERTAIN barrier for each batch,
re-checking live state
+ * and dropping any candidate revived since the scan; (3) calls {@code
plugin.deleteTopology}
+ * for the still-eligible subset; and (4) smart-finalizes the groups whose
delete succeeded.
+ * This is the cycle body {@link #startCleanupCycle} invokes on a timer. A
direct call still
+ * reads eligible batches but drops them at the {@code running} check
below unless
+ * {@link #startCleanupCycle} already flipped it true.
+ */
+ public CompletableFuture<?> runCleanupCycle() {
+ if (plugin.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+
metrics.recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_CLEANUP_CYCLE_RUNS_SENSOR_NAME);
+
+ List<CompletableFuture<Set<String>>> batchFutures =
runtime.listGroupsNeedingCleanupBatches();
+
+ // ConcurrentLinkedQueue because per-batch .handle callbacks can
append concurrently from
+ // whichever thread completed each read.
+ Queue<CompletableFuture<?>> perBatchFutures = new
ConcurrentLinkedQueue<>();
+ List<CompletableFuture<Void>> batchDoneFutures = new
ArrayList<>(batchFutures.size());
+ for (CompletableFuture<Set<String>> batchFuture : batchFutures) {
+ batchDoneFutures.add(batchFuture.handle((eligible, throwable) -> {
+ if (throwable != null) {
+ log.warn("Topology-description cleanup read failed for one
batch.", throwable);
+ return null;
+ }
+ if (eligible == null || eligible.isEmpty()) return null;
+ // Shutdown started after this read was scheduled. Skip the
plugin dispatch so we
+ // do not issue plugin.deleteTopology calls into a manager
whose plugin is about
+ // to be closed.
+ if (!running.get()) {
+ log.debug("Skipping topology-description cleanup batch:
cycle not running.");
+ return null;
+ }
+ metrics.recordSensor(
+
GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_CLEANUP_ELIGIBLE_GROUPS_SENSOR_NAME,
+ eligible.size()
+ );
+ perBatchFutures.add(cleanupBatch(eligible));
+ return null;
+ }));
+ }
+
+ return CompletableFuture.allOf(batchDoneFutures.toArray(new
CompletableFuture<?>[0]))
+ .thenCompose(__ ->
CompletableFuture.allOf(perBatchFutures.toArray(new CompletableFuture<?>[0])));
+ }
+
+ /**
+ * Drive one batch's topology cleanup for the groups the eligibility scan
returned: mark the
+ * UNCERTAIN(-2) barrier, run {@code plugin.deleteTopology} for the subset
the mark confirmed
+ * still eligible, then smart-finalize the groups whose delete succeeded.
All groups in
+ * {@code eligible} came from the same locality's read so one
mark/finalize write covers them.
+ */
+ private CompletableFuture<Void> cleanupBatch(Set<String> eligible) {
+ return runtime.markTopologyUncertainBatch(eligible)
+ .exceptionally(throwable -> {
+ // Same containment as the finalize write: one batch's routine
write failure
+ // (e.g. NOT_COORDINATOR during a move) must not fail the
whole cycle's allOf.
+ // Skipping the plugin delete is safe — the groups keep their
stored epoch and
+ // the next cycle retries.
+ log.warn("Failed to write the UNCERTAIN barrier for groups {};
skipping their "
+ + "plugin delete. The next cleanup cycle will retry.",
eligible, throwable);
+ return Set.of();
+ })
+ .thenCompose(stillEligible -> {
+ // The mark write re-checks the latest state and drops any
candidate that was
+ // revived or converted since the committed scan. Nothing to
do for a batch
+ // whose every candidate dropped out.
+ if (stillEligible.isEmpty()) return
CompletableFuture.completedFuture(null);
+ return invokeDeleteTopologies(stillEligible)
Review Comment:
If the mark write is still in flight when shutdown calls `stopCleanupCycle`,
this continuation does not re-check `running` and will invoke
`plugin.deleteTopology` after `close()` has started (the existing check only
runs before `cleanupBatch` is created). Please gate this post-mark dispatch on
the lifecycle state, and coordinate the plugin close with any
already-dispatched batch if shutdown must guarantee no plugin calls after close
begins.
--
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]
