lucasbru commented on code in PR #22622:
URL: https://github.com/apache/kafka/pull/22622#discussion_r3442406974
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -982,6 +983,52 @@ public Set<String>
streamsGroupsWithStoredTopologyDescription(
return
groupMetadataManager.streamsGroupsWithStoredTopologyDescription(groupIds,
committedOffset);
}
+ /**
+ * Return the streams groups on this shard eligible for plugin-side
topology cleanup: empty
+ * (no live members), every committed offset already past {@code
offsets.retention.ms}, and a
+ * {@code StoredDescriptionTopologyEpoch != -1}. Keyed by group id, valued
by the
+ * {@code StoredDescriptionTopologyEpoch} observed at {@code
committedOffset} — the cleanup
+ * cycle echoes that epoch back to {@link
#clearStoredDescriptionTopologyEpoch} so a
+ * concurrent {@code setTopology} that has since advanced the field cannot
be silently undone
+ * by a stale plugin delete.
+ *
+ * <p>Non-streams and missing groups are silently skipped. Per-group
errors are logged and
+ * the scan continues so one bad group cannot stall the cycle.
+ */
+ public Map<String, Integer> listStreamsGroupsNeedingTopologyCleanup(long
committedOffset) {
Review Comment:
Eligibility reads mix snapshot and live state. `maybeGroup(groupId,
committedOffset)` and `storedDescriptionTopologyEpoch(committedOffset)` are
snapshotted, but `groupMetadataManager.groupIds()` (live keySet),
`streamsGroup.isEmpty()` (live field), and
`offsetMetadataManager.allOffsetsExpired(groupId, now)` (live state +
wall-clock) are not.
The docstring above acknowledges a "concurrent setTopology" race window but
argues it's narrow because `isEmpty` gates eligibility. `isEmpty()` itself is
read live, not at `committedOffset` — so a member-join write
committed-but-not-yet-applied to the live view, or vice versa, can widen the
window beyond what the comment claims.
Either snapshot the whole eligibility check at `committedOffset` (add an
`isEmpty(long)` overload similar to the existing
`storedDescriptionTopologyEpoch(long)`), or update the docstring to say what
the actual race window looks like.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -786,6 +805,163 @@ private void
throwIfStreamsGroupTopologyDescriptionUpdateInvalid(
throwIfNull(request.topologyDescription(), "TopologyDescription can't
be null.");
}
+ /**
+ * Schedule the next topology-description cleanup tick on the broker-level
{@link Timer}.
+ * The {@link TimerTask} self-reschedules from inside its own {@code run}
so the cycle keeps
+ * firing every {@code offsets.retention.check.interval.ms} until {@code
shutdown} flips
+ * {@code isActive} to false. No-op when no plugin is configured.
+ *
+ * <p>This lifecycle lives on the service rather than on
+ * {@link StreamsGroupTopologyDescriptionManager} because the cycle drives
a runtime chain
+ * (per-shard read → per-group plugin call → conditional metadata write);
per the existing
+ * separation, the manager exposes plugin-invocation and back-off building
blocks and the
+ * service assembles the chain.
+ */
+ private void scheduleStreamsGroupTopologyCleanupCycle() {
+ if (!streamsGroupTopologyDescriptionManager.isPluginConfigured())
return;
+ if (!isActive.get()) return;
+ TimerTask task = new
TimerTask(config.offsetsRetentionCheckIntervalMs()) {
+ @Override
+ public void run() {
+ if (!isActive.get()) return;
+ try {
+ runStreamsGroupTopologyCleanupCycle();
+ } catch (Throwable t) {
+ log.warn("Unexpected error running topology-description
cleanup cycle.", t);
+ }
+ if (isActive.get()) scheduleStreamsGroupTopologyCleanupCycle();
+ }
+ };
+ streamsTopologyCleanupTask = task;
+ timer.add(task);
+ }
+
+ /**
+ * Drive one topology-description cleanup cycle: read every shard for
streams groups
+ * eligible for plugin-side cleanup (empty + all offsets expired +
storedEpoch != -1), call
+ * {@code plugin.deleteTopology} for each via the manager, then for every
group whose
+ * plugin call succeeded write a conditional metadata record that clears
+ * {@code StoredDescriptionTopologyEpoch} only if the persisted value
still matches the
+ * epoch we observed at scan time (so a concurrent {@code setTopology}
that has advanced
+ * the field is preserved). Failed plugin calls retry on the next cycle;
the next sweep
+ * then tombstones the now-empty group.
+ *
+ * <p>Single-flight: a cycle that fires while a previous one is still
settling per-group
+ * futures is dropped with a warn-level log.
+ *
+ * <p><b>Concurrent setTopology race vs plugin.deleteTopology.</b> {@code
plugin.deleteTopology}
+ * is keyed only on {@code groupId}. If a new member joins between the
+ * eligibility scan and the cycle's plugin call and pushes a fresh
topology, the plugin's
+ * row is removed regardless of the new epoch — the conditional clear
above no-ops on the
+ * metadata side, but the plugin-side data the member just wrote is gone.
A subsequent
+ * {@code describe} → {@code getTopology} returns null and surfaces {@code
NOT_STORED} with
+ * a warn log; this is the graceful-degradation path KIP-1331 accepts
under the label
+ * "plugin-side data loss". The {@code isEmpty} requirement on the scan
keeps the window
+ * narrow — concurrent setTopology requires a member to join an empty,
fully-expired group
+ * between scan and delete — and the next heartbeat at the same epoch will
not re-solicit
+ * (storedEpoch in metadata still reflects the new push), so the group
converges on
+ * NOT_STORED without churn rather than chasing the lost plugin row.
+ */
+ // Visible for testing.
+ void runStreamsGroupTopologyCleanupCycle() {
+ if (!streamsGroupTopologyDescriptionManager.isPluginConfigured())
return;
+ if (!streamsTopologyCleanupCycleInFlight.compareAndSet(false, true)) {
+ log.warn("Topology-description cleanup cycle skipped: previous
cycle is still in flight.");
+ return;
+ }
+ groupCoordinatorMetrics.recordSensor(
+
GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_CLEANUP_CYCLE_RUNS_SENSOR_NAME);
+
+ List<CompletableFuture<Map<String, Integer>>> partitionFutures =
runtime.scheduleReadAllOperation(
+ "list-streams-groups-needing-topology-cleanup",
+ GroupCoordinatorShard::listStreamsGroupsNeedingTopologyCleanup
+ );
+
+ // ConcurrentLinkedQueue because per-partition .handle callbacks can
append concurrently
+ // from whichever thread completed each runtime read.
+ Queue<CompletableFuture<?>> perGroupFutures = new
ConcurrentLinkedQueue<>();
+ List<CompletableFuture<Void>> partitionDoneFutures = new
ArrayList<>(partitionFutures.size());
+ for (CompletableFuture<Map<String, Integer>> partitionFuture :
partitionFutures) {
+ partitionDoneFutures.add(partitionFuture.handle((eligible,
throwable) -> {
+ if (throwable != null) {
+ log.warn("Topology-description cleanup read failed for one
partition.", throwable);
+ return null;
+ }
+ if (eligible == null || eligible.isEmpty()) return null;
+ groupCoordinatorMetrics.recordSensor(
+
GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_CLEANUP_ELIGIBLE_GROUPS_SENSOR_NAME,
+ eligible.size()
+ );
+ perGroupFutures.add(streamsGroupTopologyDescriptionManager
+ .invokeDeleteTopologies(eligible.keySet())
+ .thenAccept(failures -> {
+ recordPluginDeleteOutcome(eligible.size(),
failures.size());
+ eligible.forEach((groupId, expectedStoredEpoch) -> {
+ // Eligibility required the group to be empty: no
member is heartbeating
+ // anymore, so any push back-off entry is moot
regardless of plugin
+ // outcome — dropping it now avoids carrying
broker-wide state forward
+ // for an id that no live member will solicit on.
+
streamsGroupTopologyDescriptionManager.clearBackoffGroup(groupId);
Review Comment:
`clearBackoffGroup(groupId)` runs for every eligible group regardless of
plugin outcome — including the failure branch right below where the comment
explicitly says "next cycle retries". That wipes the back-off entry the plugin
manager just set/extended during the failing `invokeDeleteTopologies` call.
If a new member joins the group before the next cycle and triggers a
setTopology push, the push-path back-off check finds no entry and invokes
`plugin.setTopology` immediately, even though the plugin is in a persistently
failing state. The KIP-1331 back-off semantics get bypassed for any group the
cleanup cycle touched.
The doc above argues "no member is heartbeating anymore, so any push
back-off entry is moot" — but that assumes the group stays empty between
cycles, which isn't guaranteed (membership can change between scan and clear).
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -786,6 +805,163 @@ private void
throwIfStreamsGroupTopologyDescriptionUpdateInvalid(
throwIfNull(request.topologyDescription(), "TopologyDescription can't
be null.");
}
+ /**
+ * Schedule the next topology-description cleanup tick on the broker-level
{@link Timer}.
+ * The {@link TimerTask} self-reschedules from inside its own {@code run}
so the cycle keeps
+ * firing every {@code offsets.retention.check.interval.ms} until {@code
shutdown} flips
+ * {@code isActive} to false. No-op when no plugin is configured.
+ *
+ * <p>This lifecycle lives on the service rather than on
+ * {@link StreamsGroupTopologyDescriptionManager} because the cycle drives
a runtime chain
+ * (per-shard read → per-group plugin call → conditional metadata write);
per the existing
+ * separation, the manager exposes plugin-invocation and back-off building
blocks and the
+ * service assembles the chain.
+ */
+ private void scheduleStreamsGroupTopologyCleanupCycle() {
+ if (!streamsGroupTopologyDescriptionManager.isPluginConfigured())
return;
+ if (!isActive.get()) return;
+ TimerTask task = new
TimerTask(config.offsetsRetentionCheckIntervalMs()) {
+ @Override
+ public void run() {
+ if (!isActive.get()) return;
+ try {
+ runStreamsGroupTopologyCleanupCycle();
+ } catch (Throwable t) {
+ log.warn("Unexpected error running topology-description
cleanup cycle.", t);
+ }
+ if (isActive.get()) scheduleStreamsGroupTopologyCleanupCycle();
+ }
+ };
+ streamsTopologyCleanupTask = task;
+ timer.add(task);
+ }
+
+ /**
+ * Drive one topology-description cleanup cycle: read every shard for
streams groups
+ * eligible for plugin-side cleanup (empty + all offsets expired +
storedEpoch != -1), call
+ * {@code plugin.deleteTopology} for each via the manager, then for every
group whose
+ * plugin call succeeded write a conditional metadata record that clears
+ * {@code StoredDescriptionTopologyEpoch} only if the persisted value
still matches the
+ * epoch we observed at scan time (so a concurrent {@code setTopology}
that has advanced
+ * the field is preserved). Failed plugin calls retry on the next cycle;
the next sweep
+ * then tombstones the now-empty group.
+ *
+ * <p>Single-flight: a cycle that fires while a previous one is still
settling per-group
+ * futures is dropped with a warn-level log.
+ *
+ * <p><b>Concurrent setTopology race vs plugin.deleteTopology.</b> {@code
plugin.deleteTopology}
+ * is keyed only on {@code groupId}. If a new member joins between the
+ * eligibility scan and the cycle's plugin call and pushes a fresh
topology, the plugin's
+ * row is removed regardless of the new epoch — the conditional clear
above no-ops on the
+ * metadata side, but the plugin-side data the member just wrote is gone.
A subsequent
+ * {@code describe} → {@code getTopology} returns null and surfaces {@code
NOT_STORED} with
+ * a warn log; this is the graceful-degradation path KIP-1331 accepts
under the label
+ * "plugin-side data loss". The {@code isEmpty} requirement on the scan
keeps the window
+ * narrow — concurrent setTopology requires a member to join an empty,
fully-expired group
+ * between scan and delete — and the next heartbeat at the same epoch will
not re-solicit
+ * (storedEpoch in metadata still reflects the new push), so the group
converges on
+ * NOT_STORED without churn rather than chasing the lost plugin row.
+ */
+ // Visible for testing.
+ void runStreamsGroupTopologyCleanupCycle() {
Review Comment:
The whole cycle (single-flight `AtomicBoolean`, `scheduleReadAllOperation`
fan-out, per-group plugin invoke, per-group conditional-clear write, dual
`allOf` fan-in, self-rescheduling `TimerTask`, volatile-field snapshot for
shutdown) lives on `GroupCoordinatorService` and orchestrates a
streams-specific runtime chain.
This belongs behind `StreamsGroupTopologyDescriptionManager` — that class
already owns plugin lifecycle and back-off. The service would call one method
(`manager.runCleanupCycle(runtime)`), the timer task would live in the manager,
and the existing `clearBackoffGroup` / `invokeDeleteTopologies` would be inline
instead of cross-class calls. The hand-rolled recurring-task pattern (volatile
field + snapshot-and-cancel) also goes away — the manager can hold a stable
reference and self-arm.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -786,6 +805,163 @@ private void
throwIfStreamsGroupTopologyDescriptionUpdateInvalid(
throwIfNull(request.topologyDescription(), "TopologyDescription can't
be null.");
}
+ /**
+ * Schedule the next topology-description cleanup tick on the broker-level
{@link Timer}.
+ * The {@link TimerTask} self-reschedules from inside its own {@code run}
so the cycle keeps
+ * firing every {@code offsets.retention.check.interval.ms} until {@code
shutdown} flips
+ * {@code isActive} to false. No-op when no plugin is configured.
+ *
+ * <p>This lifecycle lives on the service rather than on
+ * {@link StreamsGroupTopologyDescriptionManager} because the cycle drives
a runtime chain
+ * (per-shard read → per-group plugin call → conditional metadata write);
per the existing
+ * separation, the manager exposes plugin-invocation and back-off building
blocks and the
+ * service assembles the chain.
+ */
+ private void scheduleStreamsGroupTopologyCleanupCycle() {
+ if (!streamsGroupTopologyDescriptionManager.isPluginConfigured())
return;
+ if (!isActive.get()) return;
+ TimerTask task = new
TimerTask(config.offsetsRetentionCheckIntervalMs()) {
+ @Override
+ public void run() {
+ if (!isActive.get()) return;
+ try {
+ runStreamsGroupTopologyCleanupCycle();
+ } catch (Throwable t) {
+ log.warn("Unexpected error running topology-description
cleanup cycle.", t);
+ }
+ if (isActive.get()) scheduleStreamsGroupTopologyCleanupCycle();
+ }
+ };
+ streamsTopologyCleanupTask = task;
+ timer.add(task);
+ }
+
+ /**
+ * Drive one topology-description cleanup cycle: read every shard for
streams groups
+ * eligible for plugin-side cleanup (empty + all offsets expired +
storedEpoch != -1), call
+ * {@code plugin.deleteTopology} for each via the manager, then for every
group whose
+ * plugin call succeeded write a conditional metadata record that clears
+ * {@code StoredDescriptionTopologyEpoch} only if the persisted value
still matches the
+ * epoch we observed at scan time (so a concurrent {@code setTopology}
that has advanced
+ * the field is preserved). Failed plugin calls retry on the next cycle;
the next sweep
+ * then tombstones the now-empty group.
+ *
+ * <p>Single-flight: a cycle that fires while a previous one is still
settling per-group
+ * futures is dropped with a warn-level log.
+ *
+ * <p><b>Concurrent setTopology race vs plugin.deleteTopology.</b> {@code
plugin.deleteTopology}
+ * is keyed only on {@code groupId}. If a new member joins between the
+ * eligibility scan and the cycle's plugin call and pushes a fresh
topology, the plugin's
+ * row is removed regardless of the new epoch — the conditional clear
above no-ops on the
+ * metadata side, but the plugin-side data the member just wrote is gone.
A subsequent
+ * {@code describe} → {@code getTopology} returns null and surfaces {@code
NOT_STORED} with
+ * a warn log; this is the graceful-degradation path KIP-1331 accepts
under the label
+ * "plugin-side data loss". The {@code isEmpty} requirement on the scan
keeps the window
+ * narrow — concurrent setTopology requires a member to join an empty,
fully-expired group
+ * between scan and delete — and the next heartbeat at the same epoch will
not re-solicit
+ * (storedEpoch in metadata still reflects the new push), so the group
converges on
+ * NOT_STORED without churn rather than chasing the lost plugin row.
+ */
+ // Visible for testing.
+ void runStreamsGroupTopologyCleanupCycle() {
+ if (!streamsGroupTopologyDescriptionManager.isPluginConfigured())
return;
+ if (!streamsTopologyCleanupCycleInFlight.compareAndSet(false, true)) {
Review Comment:
No try/finally protecting `streamsTopologyCleanupCycleInFlight`. The only
reset is the `whenComplete` at the bottom of the method. If anything between
this CAS and the chain construction throws — `recordSensor` (next line),
`scheduleReadAllOperation`, even an OOM on the ArrayList alloc — the flag stays
true forever. Outer timer-task catch reschedules normally, but every subsequent
cycle then short-circuits at the CAS and logs "previous cycle is still in
flight". Combined with the new defer gate in the sweep, streams groups with
`storedDescriptionTopologyEpoch != -1` are deferred indefinitely and never
tombstoned until broker restart.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -2639,6 +2819,13 @@ public void shutdown() {
log.info("Shutting down.");
isActive.set(false);
+ // Cancel the in-flight cleanup tick (if any) before tearing down the
runtime and the
+ // manager — the task's re-arm check will also observe isActive==false
and refuse to
+ // schedule a follow-up tick.
+ TimerTask cleanupTaskSnapshot = streamsTopologyCleanupTask;
+ if (cleanupTaskSnapshot != null) {
+ cleanupTaskSnapshot.cancel();
+ }
Utils.closeQuietly(runtime, "coordinator runtime");
Utils.closeQuietly(streamsGroupTopologyDescriptionManager, "streams
group topology description manager");
Review Comment:
`shutdown()` cancels the TimerTask snapshot then closes the runtime and the
topology-description manager. It doesn't await an in-flight cycle. `cancel()`
is non-blocking; if `runStreamsGroupTopologyCleanupCycle` is mid-flight when
shutdown fires, its plugin futures can complete *after* the manager has called
`plugin.close()`.
The `.thenAccept` then invokes `clearBackoffGroup` and
`clearStoredDescriptionTopologyEpochAsync` against a closing runtime, and
`invokeDeleteTopologies` may have already dispatched `plugin.deleteTopology`
against the now-closed plugin. Outcome is plugin-dependent (some throw on
close-then-call, some hang).
Probably want a "wait for in-flight cycle, with a small timeout" before
closing the manager — or at least an `isActive.get()` check inside the
per-group `.thenAccept` so we don't issue follow-up writes after shutdown
started.
--
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]
