Copilot commented on code in PR #22622:
URL: https://github.com/apache/kafka/pull/22622#discussion_r3441871750
##########
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) -> {
Review Comment:
`runStreamsGroupTopologyCleanupCycle` clears
`streamsTopologyCleanupCycleInFlight` after waiting on `partitionDoneFutures`
and `perGroupFutures`, but the conditional-clear writes are currently
fire-and-forget (`clearStoredDescriptionTopologyEpochAsync`) and are not
included in `perGroupFutures`. This means a subsequent cycle can start (and
re-run `plugin.deleteTopology`) before the previous cycle’s
`clear-stored-topology-epoch` writes have completed, despite the method-level
comment stating the single-flight guard covers the conditional-clear writes as
well.
Consider composing the conditional-clear write futures into the same
per-partition future (e.g., use `thenCompose` after `invokeDeleteTopologies`
and `CompletableFuture.allOf(...)` over the per-group clear writes) so the
in-flight flag is only released once the plugin calls *and* the metadata clears
have settled.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java:
##########
@@ -1034,6 +1035,60 @@ public OffsetFetchResponseData.OffsetFetchResponseGroup
fetchAllOffsets(
.setTopics(topicResponses);
}
+ /**
+ * Whether the offset committed for {@code (groupId, topic, partition)} is
currently
+ * eligible for expiration: its age exceeds {@code offsets.retention.ms}
per the group's
+ * {@link OffsetExpirationCondition}, and there is no pending
transactional offset on the
+ * partition. Shared between {@link #cleanupExpiredOffsets} (the write
path that tombstones
+ * eligible offsets) and {@link #allOffsetsExpired} (the read-only check
used by the
+ * topology-description cleanup cycle) so the two cannot drift.
+ */
+ private boolean isOffsetExpirable(
+ String groupId,
+ String topic,
+ int partition,
+ OffsetAndMetadata offsetAndMetadata,
+ OffsetExpirationCondition condition,
+ long currentTimestampMs
+ ) {
+ return condition.isOffsetExpired(offsetAndMetadata,
currentTimestampMs, config.offsetsRetentionMs())
+ && !hasPendingTransactionalOffsets(groupId, topic, partition);
+ }
+
+ /**
+ * Read-only counterpart to {@link #cleanupExpiredOffsets(String, List)}:
returns whether
+ * every committed offset for the group is currently eligible for
expiration and no pending
+ * transactional offsets remain. Used by the topology-description plugin
cleanup cycle on
+ * the eligibility read side, where the sweep must not mutate any record
but still needs to
+ * decide whether the group is fully expirable before driving a {@code
plugin.deleteTopology}.
+ */
+ public boolean allOffsetsExpired(String groupId, long currentTimestampMs) {
+ TimelineHashMap<String, TimelineHashMap<Integer, OffsetAndMetadata>>
offsetsByTopic =
+ offsets.offsetsByGroup.get(groupId);
+ if (offsetsByTopic == null) {
+ return !openTransactions.contains(groupId);
+ }
Review Comment:
`OffsetMetadataManager#allOffsetsExpired` is used from a coordinator
**read** operation
(`GroupCoordinatorShard#listStreamsGroupsNeedingTopologyCleanup(long
committedOffset)`), but it reads timeline-backed state using non-snapshot
accessors (`offsets.offsetsByGroup.get(groupId)` / `entrySet()` and
`openTransactions.contains(groupId)`), which can observe uncommitted state.
To preserve the runtime contract that read operations only consult committed
state, this method should take `committedOffset` (or otherwise be passed a
committed view) and use snapshot-aware lookups (e.g.
`offsets.offsetsByGroup.get(groupId, committedOffset)`, iterating with
`entrySet(committedOffset)`, and checking open transactions at
`committedOffset`).
--
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]
