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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
