You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lucliu1108 (via GitHub)" <gi...@apache.org> on 2026/09/23 16:17:52 UTC
[PR] KAFKA-21151: Extract streams group topology description orchestration into a shared runtime abstraction [kafka]
lucliu1108 opened a new pull request, #23566:
URL: https://github.com/apache/kafka/pull/23566
Ticket: https://issues.apache.org/jira/browse/KAFKA-21151
## Summary
This PR extracts the streams group topology description plugin's push, cleanup-cycle, and delete group orchestration out of `GroupCoordinatorService` and into `StreamsGroupTopologyDescriptionManager,` behind a new `TopologyDescriptionRuntime` interface that abstracts the underlying read/write operations. It's a behavior-preserving refactor plus test coverage: no public API or wire-protocol change, and it adds unit tests for the moved orchestration that were previously only reachable through a mocked CoordinatorRuntime.
## Files
- `TopologyDescriptionRuntime.java`: the runtime interface.
- `CoordinatorRuntimeTopologyDescriptionRuntime.java`: Group coordinator implementation.
- `StreamsGroupTopologyDescriptionManager.java`: move push, cleanup cycle, delete, and the conversion-delete building blocks original in GC service to here.
- `GroupCoordinatorService.java`" now delegates to the manager instead of calling the runtime directly.
- `StreamsGroupTopologyDescriptionManagerTest.java`: new no-op runtime test double for the existing constructor calls.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21151: Extract streams group topology description orchestration into a shared runtime abstraction [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #23566:
URL: https://github.com/apache/kafka/pull/23566
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21151: Extract streams group topology description orchestration into a shared runtime abstraction [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #23566:
URL: https://github.com/apache/kafka/pull/23566#discussion_r4100085103
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -1355,7 +1089,7 @@ private CompletableFuture<Void> cleanupTopologyBeforeConversion(
failJoinRetriably(request, responseFuture);
return CompletableFuture.completedFuture(null);
}
- return markTopologyUncertainAsync(tp, request.groupId(), false)
+ return streamsGroupTopologyDescriptionManager.markTopologyUncertain(request.groupId(), false)
Review Comment:
Fix: Created a new `StreamsGroupTopologyDescription#cleanupTopologyBeforeConversion` method to gather all those steps in, and return a boolean to indicate if the cleanup succeeded. If not, retry in GC service.
Also marked those methods private.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21151: Extract streams group topology description orchestration into a shared runtime abstraction [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on PR #23566:
URL: https://github.com/apache/kafka/pull/23566#issuecomment-5827449473
Thanks @lucasbru , for related tests isolation, i created this ticket https://issues.apache.org/jira/browse/KAFKA-21162 to follow up.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21151: Extract streams group topology description orchestration into a shared runtime abstraction [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #23566:
URL: https://github.com/apache/kafka/pull/23566#discussion_r4094305006
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java:
##########
@@ -275,6 +458,33 @@ public StreamsGroupHeartbeatResult maybeSetTopologyDescriptionRequired(
return result;
}
+ /**
+ * Decorates a streams group heartbeat's result with {@link #maybeSetTopologyDescriptionRequired},
+ * once the underlying write completes. No-ops when no plugin is configured, without
+ * scheduling the extra stage. If decoration itself fails (e.g. because of an unexpected
+ * response shape), the heartbeat has already committed durably, so the failure is logged and
+ * the committed result is returned unmodified rather than translated into an error; that
+ * would mask a successful broker-side state change behind a client-visible failure.
+ */
+ public CompletableFuture<StreamsGroupHeartbeatResult> decorateHeartbeatResult(
+ CompletableFuture<StreamsGroupHeartbeatResult> heartbeat,
+ String groupId,
+ int apiVersion,
+ int memberEpoch
+ ) {
+ if (!isPluginConfigured()) return heartbeat;
+ return heartbeat.thenApply(result -> {
+ try {
+ return maybeSetTopologyDescriptionRequired(result, groupId, apiVersion, memberEpoch);
Review Comment:
Now that this wraps it, GroupCoordinatorService doesn't call maybeSetTopologyDescriptionRequired directly anymore, so it could drop to private. Same for invokeSetTopology and completeEpochWrite in pushTopology above - both were public building blocks for the service to compose before this PR, but now that pushTopology composes them itself, nothing outside this class calls them either.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java:
##########
@@ -134,22 +212,25 @@ public void startCleanupCycle(
scheduleNextTick(timer, cleanupCheckIntervalMs, cycleSupplier);
}
- /**
- * Stop the cleanup cycle and release plugin-side resources. Flips {@code running}
- * false (so the next timer tick refuses to fire) and cancels the currently-scheduled
- * tick, then closes the plugin. Called by {@code GroupCoordinatorService.shutdown}
- * before the runtime is closed, so writes already scheduled by the previous tick
- * drain through their own futures rather than racing the runtime tear-down.
- */
- @Override
- public void close() throws Exception {
+ /** Stops the cleanup cycle without touching the plugin; see {@link #close}. */
+ public void stopCleanupCycle() {
if (running.compareAndSet(true, false)) {
TimerTask snapshot = scheduledTask;
if (snapshot != null) {
snapshot.cancel();
}
}
- if (plugin.isPresent()) {
+ }
+
+ /**
+ * Stops the cleanup cycle and closes the plugin. Called by
+ * {@code GroupCoordinatorService.shutdown} after {@code isActive} flips false, since request
+ * handlers gate on {@code isActive} once and don't re-check it before reaching the plugin.
+ */
+ @Override
+ public void close() throws Exception {
+ stopCleanupCycle();
+ if (closed.compareAndSet(false, true) && plugin.isPresent()) {
Review Comment:
Two gaps that are actually new with this PR and have no coverage anywhere: (1) the closed guard here - no test calls close()/shutdown() twice to check plugin.close() only fires once; before this PR there was no guard to test, now there's a real branch. (2) stopCleanupCycle() itself - it's only ever exercised indirectly through shutdown()'s end-to-end timer-size check, never called directly to verify its own no-op-when-not-running behavior. Both are cheap to add now that StreamsGroupTopologyDescriptionManagerTest exists - e.g. build a manager, call close() twice, assert plugin.close() invoked once; and call stopCleanupCycle() before/after startCleanupCycle() and check the scheduled task.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -1355,7 +1089,7 @@ private CompletableFuture<Void> cleanupTopologyBeforeConversion(
failJoinRetriably(request, responseFuture);
return CompletableFuture.completedFuture(null);
}
- return markTopologyUncertainAsync(tp, request.groupId(), false)
+ return streamsGroupTopologyDescriptionManager.markTopologyUncertain(request.groupId(), false)
Review Comment:
This whole method already routes every topology-specific step through `streamsGroupTopologyDescriptionManager` (isConversionDeleteThrottled, markTopologyUncertain, invokeDeleteTopologies, recordPluginDeleteOutcome, throttleConversionDelete, clearBackoffGroup, finalizeAfterDelete). Could the throttle -> mark -> delete -> record -> throttle-or-clear-backoff sequence move onto the manager as one method returning a boolean (cleaned up or not)? All three failure branches here do the same thing (failJoinRetriably), so this would collapse to something like:
```java
streamsGroupTopologyDescriptionManager.cleanupTopologyBeforeConversion(request.groupId())
.thenCompose(cleaned -> {
if (!cleaned) {
failJoinRetriably(request, responseFuture);
return CompletableFuture.completedFuture(null);
}
return runClassicGroupJoin(context, request, responseFuture, tp, true)
.thenCompose(__ -> streamsGroupTopologyDescriptionManager.finalizeAfterDelete(request.groupId()));
});
```
`runClassicGroupJoin`/`failJoinRetriably` would stay here since they need the raw runtime and the JoinGroup wire types, but the rest looks like it belongs with the orchestration that already moved. Would also let `isConversionDeleteThrottled`, `throttleConversionDelete`, `markTopologyUncertain`, and `clearBackoffGroup` drop to private on the manager, since nothing outside it would call them anymore.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21151: Extract streams group topology description orchestration into a shared runtime abstraction [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #23566:
URL: https://github.com/apache/kafka/pull/23566#discussion_r4100066681
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java:
##########
@@ -134,22 +212,25 @@ public void startCleanupCycle(
scheduleNextTick(timer, cleanupCheckIntervalMs, cycleSupplier);
}
- /**
- * Stop the cleanup cycle and release plugin-side resources. Flips {@code running}
- * false (so the next timer tick refuses to fire) and cancels the currently-scheduled
- * tick, then closes the plugin. Called by {@code GroupCoordinatorService.shutdown}
- * before the runtime is closed, so writes already scheduled by the previous tick
- * drain through their own futures rather than racing the runtime tear-down.
- */
- @Override
- public void close() throws Exception {
+ /** Stops the cleanup cycle without touching the plugin; see {@link #close}. */
+ public void stopCleanupCycle() {
if (running.compareAndSet(true, false)) {
TimerTask snapshot = scheduledTask;
if (snapshot != null) {
snapshot.cancel();
}
}
- if (plugin.isPresent()) {
+ }
+
+ /**
+ * Stops the cleanup cycle and closes the plugin. Called by
+ * {@code GroupCoordinatorService.shutdown} after {@code isActive} flips false, since request
+ * handlers gate on {@code isActive} once and don't re-check it before reaching the plugin.
+ */
+ @Override
+ public void close() throws Exception {
+ stopCleanupCycle();
+ if (closed.compareAndSet(false, true) && plugin.isPresent()) {
Review Comment:
Added 3 new tests:
- `testCloseInvokesPluginCloseOnlyOnce`: closes manager twice, each time the `plugin.close()` is triggered it increment a counter. Validate that counter is 1.
- `testStopCleanupCycleNoOpWhenNotRunning`: don't spin up the manager and call `manager#stopCleanupCycle()` , verify the manager is not running and no scheduledCleanupTask available.
- `testStopCleanupCycleCancelsScheduledTask`: spin up the manager, validate there's `scheduledCleanupTask` running. Then call `stopCleanupCycle`, verify that the manager isn't running and the cleanupTask is cancelled.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21151: Extract streams group topology description orchestration into a shared runtime abstraction [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #23566:
URL: https://github.com/apache/kafka/pull/23566#discussion_r4100068251
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java:
##########
@@ -275,6 +458,33 @@ public StreamsGroupHeartbeatResult maybeSetTopologyDescriptionRequired(
return result;
}
+ /**
+ * Decorates a streams group heartbeat's result with {@link #maybeSetTopologyDescriptionRequired},
+ * once the underlying write completes. No-ops when no plugin is configured, without
+ * scheduling the extra stage. If decoration itself fails (e.g. because of an unexpected
+ * response shape), the heartbeat has already committed durably, so the failure is logged and
+ * the committed result is returned unmodified rather than translated into an error; that
+ * would mask a successful broker-side state change behind a client-visible failure.
+ */
+ public CompletableFuture<StreamsGroupHeartbeatResult> decorateHeartbeatResult(
+ CompletableFuture<StreamsGroupHeartbeatResult> heartbeat,
+ String groupId,
+ int apiVersion,
+ int memberEpoch
+ ) {
+ if (!isPluginConfigured()) return heartbeat;
+ return heartbeat.thenApply(result -> {
+ try {
+ return maybeSetTopologyDescriptionRequired(result, groupId, apiVersion, memberEpoch);
Review Comment:
Fix: change all 3 methods to private.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21151: Extract streams group topology description orchestration into a shared runtime abstraction [kafka]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org