This is an automated email from the ASF dual-hosted git repository.
lucasbru pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new d4cd81a5ef2 KAFKA-20696: Periodic plugin.deleteTopology cleanup for
naturally-expired streams groups (#22622)
d4cd81a5ef2 is described below
commit d4cd81a5ef2d88db7c2851c17f4cb76107cee2ef
Author: TengYao Chi <[email protected]>
AuthorDate: Wed Jun 24 11:02:42 2026 +0100
KAFKA-20696: Periodic plugin.deleteTopology cleanup for naturally-expired
streams groups (#22622)
JIRA: KAFKA-20696 This PR is a part of KIP-1331
Cycle (broker-level, on GroupCoordinatorService):
- A self-rescheduling TimerTask runs at
offsets.retention.check.interval.ms,
gated on isActive + manager.isPluginConfigured(). Single-flight
guarded so a
cycle that fires while the previous one is still settling per-group
futures
is dropped with a warn log.
- Each tick calls runtime.scheduleReadAllOperation to fan out an
eligibility
query to every hosted shard, then fans plugin.deleteTopology out via
the
existing manager.invokeDeleteTopologies building block. The manager
itself
is unchanged — it keeps the (plugin, time) shape KAFKA-20623-4 settled
on
and exposes only plugin-invocation + back-off mutations; chain
assembly
lives on the service.
Eligibility + gating (per-shard):
- New GroupCoordinatorShard.listStreamsGroupsNeedingTopologyCleanup
returns
group ids that are STREAMS + isEmpty + storedDescriptionTopologyEpoch
!= -1
+ offsets all past retention, paired with the observed storedEpoch.
The
scan uses a non-throwing GroupMetadataManager.maybeGroup lookup to
avoid
per-non-streams-group exception cost on plugin-enabled brokers running
mixed groups; per-group errors are logged and the scan continues.
- cleanupGroupMetadata now defers maybeDeleteGroup for STREAMS groups
with
storedEpoch != -1, but only when a plugin is configured on this broker
—
otherwise no cycle would ever clear the field and the gate would block
natural expiration indefinitely. The cast is factored into a named
helper
so the short-circuit precondition for (StreamsGroup) sits next to the
cast.
- After a successful plugin.deleteTopology, the cycle writes a
conditional
GroupMetadataManager.clearStoredDescriptionTopologyEpoch(groupId,
expectedStoredEpoch) — only clears when the persisted value still
equals
the epoch we observed at scan time, so a concurrent setTopology that
has
advanced storedEpoch is preserved. Failed plugin calls retry on the
next
cycle.
- The empty-group → concurrent-setTopology → cycle-delete race is
documented
on runStreamsGroupTopologyCleanupCycle: plugin.deleteTopology keys
only on
groupId per KIP appendix, so a new member that pushes between scan and
delete loses its plugin row; describe surfaces NOT_STORED + WARN,
which is
the graceful-degradation path KIP-1331 explicitly accepts as
"plugin-side
data loss".
OffsetMetadataManager:
- Adds a read-only allOffsetsExpired(groupId, currentTimestampMs) used
by the
shard scan. The per-offset expirability check is factored into a
private
isOffsetExpirable helper shared with cleanupExpiredOffsets so the read
and
write paths cannot drift.
Metrics (KIP-1331):
- streams-group-topology-description-cleanup-cycle-{rate,count}
- streams-group-topology-description-cleanup-eligible-{rate,count}
- streams-group-topology-description-delete-{success,error}-{rate,count}
Reviewers: Lucas Brutschy <[email protected]>
---
.../coordinator/group/GroupCoordinatorConfig.java | 9 +
.../coordinator/group/GroupCoordinatorService.java | 167 +++++++++++-
.../coordinator/group/GroupCoordinatorShard.java | 88 ++++++-
.../coordinator/group/GroupMetadataManager.java | 51 ++++
.../coordinator/group/OffsetMetadataManager.java | 83 +++++-
.../group/metrics/GroupCoordinatorMetrics.java | 79 +++++-
.../coordinator/group/streams/StreamsGroup.java | 12 +
.../StreamsGroupTopologyDescriptionManager.java | 147 ++++++++++-
...pCoordinatorServiceTopologyDescriptionTest.java | 289 +++++++++++++++++++++
.../group/GroupCoordinatorShardTest.java | 272 +++++++++++++++++++
.../group/GroupMetadataManagerTest.java | 49 ++++
.../group/OffsetMetadataManagerTest.java | 152 +++++++++++
.../group/metrics/GroupCoordinatorMetricsTest.java | 8 +
13 files changed, 1376 insertions(+), 30 deletions(-)
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java
index af4aaf2169c..31454863565 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java
@@ -918,6 +918,15 @@ public class GroupCoordinatorConfig {
overrides);
}
+ /**
+ * Whether a topology-description plugin class is set on this broker.
Checked from per-shard
+ * paths (the {@code cleanupGroupMetadata} gate) that cannot reach the
broker-level manager
+ * holding the plugin reference. Does not instantiate the plugin; just
inspects the config.
+ */
+ public boolean isStreamsGroupTopologyDescriptionPluginConfigured() {
+ return
config.getClass(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG) != null;
+ }
+
/**
* The number of threads or event loops running.
*/
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
index cad84307b08..dfe017f8108 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
@@ -143,9 +143,11 @@ import java.util.Objects;
import java.util.Optional;
import java.util.OptionalInt;
import java.util.Properties;
+import java.util.Queue;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.IntSupplier;
@@ -363,13 +365,19 @@ public class GroupCoordinatorService implements
GroupCoordinator {
private final PartitionMetadataClient partitionMetadataClient;
/**
- * The broker-level component that owns the streams-group topology
description plugin
- * (KIP-1331): plugin reference, per-group push back-off, and the three
entry points
- * the service delegates into — heartbeat post-processing, the push RPC,
and the
- * pre-tombstone hook on DeleteGroups.
+ * The broker-level component that owns the streams-group topology
description plugin:
+ * plugin reference, per-group push back-off, the entry points the service
delegates
+ * into (heartbeat post-processing, the push RPC, the pre-tombstone hook on
+ * DeleteGroups), and the periodic plugin-row cleanup cycle for
naturally-expired
+ * streams groups.
*/
private final StreamsGroupTopologyDescriptionManager
streamsGroupTopologyDescriptionManager;
+ // Visible for testing.
+ StreamsGroupTopologyDescriptionManager
streamsGroupTopologyDescriptionManager() {
+ return streamsGroupTopologyDescriptionManager;
+ }
+
/**
* The number of partitions of the __consumer_offsets topics. This is
provided
* when the component is started.
@@ -786,6 +794,144 @@ public class GroupCoordinatorService implements
GroupCoordinator {
throwIfNull(request.topologyDescription(), "TopologyDescription can't
be null.");
}
+ /**
+ * Build 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>The single-flight guard, periodic timer scheduling, and {@code
running} flag live
+ * on {@link StreamsGroupTopologyDescriptionManager#startCleanupCycle};
this method is
+ * the cycle body it invokes, returning a future that the manager joins to
release the
+ * in-flight flag.
+ *
+ * <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 accepted 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.
+ CompletableFuture<?> runOneStreamsTopologyCleanupCycle() {
+ if (!streamsGroupTopologyDescriptionManager.isPluginConfigured()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ 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;
+ // Shutdown started after the per-partition 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 (!isActive.get()) return null;
+ groupCoordinatorMetrics.recordSensor(
+
GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_CLEANUP_ELIGIBLE_GROUPS_SENSOR_NAME,
+ eligible.size()
+ );
+ perGroupFutures.add(streamsGroupTopologyDescriptionManager
+ .invokeDeleteTopologies(eligible.keySet())
+ .thenCompose(failures -> {
+ recordPluginDeleteOutcome(eligible.size(),
failures.size());
+ // Shutdown can have started between the plugin call
and the
+ // follow-up writes. Skip the conditional clears so we
do not
+ // schedule writes against a runtime that is being
closed.
+ if (!isActive.get()) return
CompletableFuture.completedFuture(null);
+ List<CompletableFuture<Void>> clearFutures = new
ArrayList<>(eligible.size());
+ eligible.forEach((groupId, expectedStoredEpoch) -> {
+ if (failures.containsKey(groupId)) {
+ // Plugin failed: leave both stored epoch and
the push-path
+ // back-off in place. Eligibility's "group is
empty" snapshot
+ // only held at scan time; a member can rejoin
between scan
+ // and now, and the existing back-off
correctly throttles their
+ // set-topology attempt against the
still-broken plugin
+ // instead of letting it re-attack at
attempts=0 every join.
+ return;
+ }
+ // Plugin succeeded; the group will be tombstoned
in the next sweep
+ // once the stored epoch is cleared. Drop the
broker-wide back-off
+ // entry — it is no longer load-bearing for any
future state of
+ // this groupId. A member that re-creates the same
id afterwards
+ // is a fresh lifecycle and will arm a fresh
back-off chain.
+
streamsGroupTopologyDescriptionManager.clearBackoffGroup(groupId);
+
clearFutures.add(clearStoredDescriptionTopologyEpochAsync(groupId,
expectedStoredEpoch));
+ });
+ return
CompletableFuture.allOf(clearFutures.toArray(new CompletableFuture<?>[0]));
+ }));
+ return null;
+ }));
+ }
+
+ return CompletableFuture.allOf(partitionDoneFutures.toArray(new
CompletableFuture<?>[0]))
+ .thenCompose(__ ->
CompletableFuture.allOf(perGroupFutures.toArray(new CompletableFuture<?>[0])));
+ }
+
+ /**
+ * Record per-call outcomes from a batched {@code plugin.deleteTopology}
invocation
+ * against the shared {@code delete-success} / {@code delete-error}
sensors. Used by
+ * the periodic cleanup cycle and the explicit {@code DeleteGroups} flow
so a single
+ * pair of meters tracks every {@code plugin.deleteTopology} the broker
drives,
+ * regardless of trigger.
+ */
+ private void recordPluginDeleteOutcome(int attempted, int errors) {
+ int successes = attempted - errors;
+ if (successes > 0) {
+ groupCoordinatorMetrics.recordSensor(
+
GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_DELETE_SUCCESS_SENSOR_NAME,
successes);
+ }
+ if (errors > 0) {
+ groupCoordinatorMetrics.recordSensor(
+
GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_DELETE_ERROR_SENSOR_NAME,
errors);
+ }
+ }
+
+ /**
+ * Conditional metadata write that clears {@code
StoredDescriptionTopologyEpoch} for
+ * {@code groupId} only when the persisted value still equals {@code
expectedStoredEpoch}.
+ * Mismatches and missing groups are silently ignored by the shard-side
method. Runtime
+ * write failures (NOT_COORDINATOR etc.) are logged here and swallowed so
a single failed
+ * write does not poison the cycle's allOf — the next cycle will retry
naturally because
+ * the persisted storedEpoch is still non-default.
+ */
+ private CompletableFuture<Void>
clearStoredDescriptionTopologyEpochAsync(String groupId, int
expectedStoredEpoch) {
+ return runtime.<Void>scheduleWriteOperation(
+ "clear-stored-topology-epoch",
+ topicPartitionFor(groupId),
+ coordinator ->
coordinator.clearStoredDescriptionTopologyEpoch(groupId, expectedStoredEpoch)
+ ).handle((__, throwable) -> {
+ if (throwable != null) {
+ log.warn("Failed to clear StoredDescriptionTopologyEpoch for
group {}; the next cleanup cycle will retry.",
+ groupId, throwable);
+ }
+ return null;
+ });
+ }
+
/**
* Validates the ShareGroupHeartbeat request.
*
@@ -2627,6 +2773,13 @@ public class GroupCoordinatorService implements
GroupCoordinator {
log.info("Starting up.");
numPartitions = groupMetadataTopicPartitionCount.getAsInt();
isActive.set(true);
+ // Arm the periodic topology-description cleanup cycle on the manager;
no-op when no
+ // plugin is configured. The manager owns the timer + single-flight
harness; the
+ // cycle body lives here on the service via
runOneStreamsTopologyCleanupCycle.
+ streamsGroupTopologyDescriptionManager.startCleanupCycle(
+ timer,
+ config.offsetsRetentionCheckIntervalMs(),
+ this::runOneStreamsTopologyCleanupCycle);
log.info("Startup complete.");
}
@@ -2642,8 +2795,12 @@ public class GroupCoordinatorService implements
GroupCoordinator {
log.info("Shutting down.");
isActive.set(false);
- Utils.closeQuietly(runtime, "coordinator runtime");
+ // Close the topology-description manager before the runtime so that
its cycle's
+ // running flag flips false and the scheduled tick is cancelled while
the runtime is
+ // still alive — writes already scheduled before the flip drain
through their own
+ // futures rather than racing the runtime tear-down.
Utils.closeQuietly(streamsGroupTopologyDescriptionManager, "streams
group topology description manager");
+ Utils.closeQuietly(runtime, "coordinator runtime");
Utils.closeQuietly(groupCoordinatorMetrics, "group coordinator
metrics");
Utils.closeQuietly(groupConfigManager, "group config manager");
log.info("Shutdown complete.");
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
index b3dc8efa6c4..b79f7ebe3a4 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
@@ -120,6 +120,7 @@ import
org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetricsShard;
import org.apache.kafka.coordinator.group.modern.share.ShareGroup;
+import org.apache.kafka.coordinator.group.streams.StreamsGroup;
import org.apache.kafka.coordinator.group.streams.StreamsGroupDescribeResult;
import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
import org.apache.kafka.server.authorizer.AuthorizableRequestContext;
@@ -982,6 +983,60 @@ public class GroupCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
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>Every state lookup goes through the snapshot at {@code
committedOffset}: the iterated
+ * group-id set, the per-group resolution, the {@code EMPTY}-state check,
the stored
+ * topology epoch, and the per-offset retention check inside
+ * {@link OffsetMetadataManager#allOffsetsExpired}. The only non-snapshot
input is
+ * {@code now} from the wall clock, which is unavoidable for "offset has
aged past
+ * retention" and is captured once at the top of the scan so every group
is compared
+ * against the same instant.
+ *
+ * <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) {
+ long now = time.milliseconds();
+ Map<String, Integer> eligible = new HashMap<>();
+ for (String groupId : groupMetadataManager.groupIds(committedOffset)) {
+ try {
+ Group group = groupMetadataManager.maybeGroup(groupId,
committedOffset);
+ if (group == null || group.type() != Group.GroupType.STREAMS)
continue;
+ StreamsGroup streamsGroup = (StreamsGroup) group;
+ if (!streamsGroup.isEmpty(committedOffset)) continue;
+ int storedEpoch =
streamsGroup.storedDescriptionTopologyEpoch(committedOffset);
+ if (storedEpoch == -1) continue;
+ if (!offsetMetadataManager.allOffsetsExpired(groupId, now,
committedOffset)) continue;
+ eligible.put(groupId, storedEpoch);
+ } catch (Throwable t) {
+ // One bad group must not abort the whole scan; the next cycle
retries.
+ log.warn("Unexpected error scanning streams group {} for
topology cleanup; skipping.", groupId, t);
+ }
+ }
+ return eligible;
+ }
+
+ /**
+ * Clear {@code StoredDescriptionTopologyEpoch} to {@code -1} for {@code
groupId}, but only
+ * when the persisted value still equals {@code expectedStoredEpoch}.
Called from the
+ * topology-description cleanup cycle so a concurrent {@code setTopology}
that has advanced
+ * the field is preserved.
+ */
+ public CoordinatorResult<Void, CoordinatorRecord>
clearStoredDescriptionTopologyEpoch(
+ String groupId,
+ int expectedStoredEpoch
+ ) {
+ return
groupMetadataManager.clearStoredDescriptionTopologyEpoch(groupId,
expectedStoredEpoch);
+ }
+
/**
* Handles a ShareGroupDescribe request.
*
@@ -1054,15 +1109,15 @@ public class GroupCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
*/
public CoordinatorResult<Void, CoordinatorRecord> cleanupGroupMetadata() {
long startMs = time.milliseconds();
+ boolean topologyPluginConfigured =
config.isStreamsGroupTopologyDescriptionPluginConfigured();
List<CoordinatorRecord> records = new ArrayList<>();
groupMetadataManager.groupIds().forEach(groupId -> {
Group group = groupMetadataManager.group(groupId);
- if (group.shouldExpire()) {
- boolean allOffsetsExpired =
offsetMetadataManager.cleanupExpiredOffsets(groupId, records);
- if (allOffsetsExpired) {
- groupMetadataManager.maybeDeleteGroup(groupId, records);
- }
- }
+ if (!group.shouldExpire()) return;
+ boolean allOffsetsExpired =
offsetMetadataManager.cleanupExpiredOffsets(groupId, records);
+ if (!allOffsetsExpired) return;
+ if
(deferStreamsGroupTombstoneForPluginCleanup(topologyPluginConfigured, group))
return;
+ groupMetadataManager.maybeDeleteGroup(groupId, records);
});
if (!records.isEmpty()) {
@@ -1075,6 +1130,27 @@ public class GroupCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
return new CoordinatorResult<>(records, false);
}
+ /**
+ * Decide whether the natural-expiration sweep must defer tombstoning
{@code group} so the
+ * broker-level topology-description cleanup cycle can drive {@code
plugin.deleteTopology}
+ * and clear {@code StoredDescriptionTopologyEpoch} first. Returns true
only when all of:
+ * a plugin is configured on this broker (otherwise no cycle would ever
clear the field
+ * and the gate would prevent natural expiration indefinitely), the group
is a streams
+ * group, and its {@code StoredDescriptionTopologyEpoch} is not the {@code
-1} default.
+ *
+ * <p>The {@code (StreamsGroup) group} cast is safe because the {@code
group.type() == STREAMS}
+ * check precedes it via short-circuit evaluation; pulling this out of the
sweep lambda
+ * keeps the gate's intent and the cast's precondition next to each other.
+ */
+ private static boolean deferStreamsGroupTombstoneForPluginCleanup(
+ boolean topologyPluginConfigured,
+ Group group
+ ) {
+ return topologyPluginConfigured
+ && group.type() == Group.GroupType.STREAMS
+ && ((StreamsGroup) group).storedDescriptionTopologyEpoch() != -1;
+ }
+
/**
* Schedule the group/offsets expiration job. If any exceptions are thrown
above, the timer will retry.
*/
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
index b0136ca3d95..08a0abfda04 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
@@ -627,6 +627,16 @@ public class GroupMetadataManager {
return group;
}
+ /**
+ * Non-throwing variant of {@link #group(String, long)}: returns {@code
null} if no group
+ * with the given id exists at {@code committedOffset}. Used by scans
(e.g. the topology-
+ * description cleanup cycle) where a missing group is a normal
continue-the-scan condition
+ * rather than an error worth a {@link GroupIdNotFoundException} cost per
iteration.
+ */
+ public Group maybeGroup(String groupId, long committedOffset) {
+ return groups.get(groupId, committedOffset);
+ }
+
/**
* Get the Group List.
*
@@ -8500,6 +8510,37 @@ public class GroupMetadataManager {
return new CoordinatorResult<>(List.of(record), null);
}
+ /**
+ * Clear {@code StoredDescriptionTopologyEpoch} to {@code -1} only when
the group's stored
+ * epoch still equals {@code expectedStoredEpoch} (the value observed at
the start of the
+ * cleanup cycle). A concurrent {@code setTopology} that advanced the
epoch in the
+ * meantime is preserved — the next cycle will pick up the new state.
Missing groups,
+ * non-streams groups, and mismatched epochs all yield an empty record
list so the cycle's
+ * downstream tombstone pass treats them as no-ops rather than errors.
+ */
+ public CoordinatorResult<Void, CoordinatorRecord>
clearStoredDescriptionTopologyEpoch(
+ String groupId,
+ int expectedStoredEpoch
+ ) {
+ Group group = groups.get(groupId);
+ if (!(group instanceof StreamsGroup streamsGroup)) {
+ return new CoordinatorResult<>(List.of());
+ }
+ if (streamsGroup.storedDescriptionTopologyEpoch() !=
expectedStoredEpoch) {
+ return new CoordinatorResult<>(List.of());
+ }
+ CoordinatorRecord record = newStreamsGroupMetadataRecord(
+ groupId,
+ streamsGroup.groupEpoch(),
+ streamsGroup.metadataHash(),
+ streamsGroup.validatedTopologyEpoch(),
+ streamsGroup.lastAssignmentConfigs(),
+ -1,
+ streamsGroup.failedDescriptionTopologyEpoch()
+ );
+ return new CoordinatorResult<>(List.of(record), null);
+ }
+
/**
* Validates that (1) the instance id exists and is mapped to the member id
* if the group instance id is provided; and (2) the member id exists in
the group.
@@ -9295,6 +9336,16 @@ public class GroupMetadataManager {
return Collections.unmodifiableSet(this.groups.keySet());
}
+ /**
+ * Snapshot-aware counterpart to {@link #groupIds()}: returns the set of
group ids
+ * present at {@code committedOffset}. Used by read operations whose
entire eligibility
+ * decision must be reproducible from a single committed snapshot (e.g.
the topology
+ * cleanup scan in {@link GroupCoordinatorShard}).
+ */
+ public Set<String> groupIds(long committedOffset) {
+ return
Collections.unmodifiableSet(this.groups.keySet(committedOffset));
+ }
+
// Visible for testing
Map<String, Long> topicHashCache() {
return Collections.unmodifiableMap(this.topicHashCache);
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java
index e30b036c66e..1465af9e8e4 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java
@@ -57,6 +57,7 @@ import org.slf4j.Logger;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.OptionalLong;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -258,6 +259,16 @@ public class OffsetMetadataManager {
return openTransactionsByGroup.containsKey(groupId);
}
+ /**
+ * Snapshot-aware overload of {@link #contains(String)}: returns
{@code true} if the
+ * given group had any pending transactional offsets at {@code
committedOffset}. Used
+ * by read operations that must only observe committed state (e.g. the
topology
+ * cleanup cycle's eligibility scan).
+ */
+ private boolean contains(String groupId, long committedOffset) {
+ return openTransactionsByGroup.containsKey(groupId,
committedOffset);
+ }
+
/**
* Returns {@code true} if the given group has any pending
transactional offsets for the given topic and partition.
*
@@ -271,6 +282,21 @@ public class OffsetMetadataManager {
return openTransactions != null;
}
+ /**
+ * Snapshot-aware overload of {@link #contains(String, String, int)}:
returns
+ * {@code true} if the given group had any pending transactional
offsets for the
+ * given topic and partition at {@code committedOffset}.
+ */
+ private boolean contains(String groupId, String topic, int partition,
long committedOffset) {
+ TimelineHashMap<String, TimelineHashMap<Integer,
TimelineHashSet<Long>>> openTransactionsByTopic =
+ openTransactionsByGroup.get(groupId, committedOffset);
+ if (openTransactionsByTopic == null) return false;
+ TimelineHashMap<Integer, TimelineHashSet<Long>>
openTransactionsByPartition =
+ openTransactionsByTopic.get(topic, committedOffset);
+ if (openTransactionsByPartition == null) return false;
+ return openTransactionsByPartition.containsKey(partition,
committedOffset);
+ }
+
/**
* Performs the given action for each partition with a pending
transactional offset for the given group.
*
@@ -1034,6 +1060,58 @@ public class OffsetMetadataManager {
.setTopics(topicResponses);
}
+ /**
+ * 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}.
+ *
+ * <p>{@code committedOffset} is the snapshot point the runtime hands to
the read operation
+ * that calls this method. Every timeline-backed lookup in here uses that
snapshot — the
+ * runtime contract is that read operations only observe committed state,
so a concurrent
+ * uncommitted offset commit or pending-transaction record must not flip
the eligibility
+ * outcome on us.
+ */
+ public boolean allOffsetsExpired(String groupId, long currentTimestampMs,
long committedOffset) {
+ TimelineHashMap<String, TimelineHashMap<Integer, OffsetAndMetadata>>
offsetsByTopic =
+ offsets.offsetsByGroup.get(groupId, committedOffset);
+ if (offsetsByTopic == null) {
+ return !openTransactions.contains(groupId, committedOffset);
+ }
+ Group group;
+ try {
+ group = groupMetadataManager.group(groupId, committedOffset);
+ } catch (GroupIdNotFoundException e) {
+ // The group disappeared between the caller's existence check and
this lookup at
+ // the same snapshot — it is not eligible for plugin cleanup, the
next sweep will
+ // pick this up naturally.
+ return false;
+ }
+ Optional<OffsetExpirationCondition> offsetExpirationCondition =
group.offsetExpirationCondition();
+ if (offsetExpirationCondition.isEmpty()) {
+ return false;
+ }
+ OffsetExpirationCondition condition = offsetExpirationCondition.get();
+ for (Map.Entry<String, TimelineHashMap<Integer, OffsetAndMetadata>>
topicEntry
+ : offsetsByTopic.entrySet(committedOffset)) {
+ String topic = topicEntry.getKey();
+ if (group.isSubscribedToTopic(topic)) {
+ return false;
+ }
+ for (Map.Entry<Integer, OffsetAndMetadata> partitionEntry
+ : topicEntry.getValue().entrySet(committedOffset)) {
+ int partition = partitionEntry.getKey();
+ OffsetAndMetadata offsetAndMetadata =
partitionEntry.getValue();
+ if (!condition.isOffsetExpired(offsetAndMetadata,
currentTimestampMs, config.offsetsRetentionMs())
+ || openTransactions.contains(groupId, topic, partition,
committedOffset)) {
+ return false;
+ }
+ }
+ }
+ return !openTransactions.contains(groupId, committedOffset);
+ }
+
/**
* Remove expired offsets for the given group.
*
@@ -1065,9 +1143,8 @@ public class OffsetMetadataManager {
offsetsByTopic.forEach((topic, partitions) -> {
if (!group.isSubscribedToTopic(topic)) {
partitions.forEach((partition, offsetAndMetadata) -> {
- // We don't expire the offset yet if there is a pending
transactional offset for the partition.
- if (condition.isOffsetExpired(offsetAndMetadata,
currentTimestampMs, config.offsetsRetentionMs()) &&
- !hasPendingTransactionalOffsets(groupId, topic,
partition)) {
+ if (condition.isOffsetExpired(offsetAndMetadata,
currentTimestampMs, config.offsetsRetentionMs())
+ && !hasPendingTransactionalOffsets(groupId, topic,
partition)) {
appendOffsetCommitTombstone(groupId, topic, partition,
records);
log.debug("[GroupId {}] Expired offset for
partition={}-{}", groupId, topic, partition);
} else {
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java
index af67090bd02..185ea52d83a 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java
@@ -122,6 +122,10 @@ public class GroupCoordinatorMetrics extends
CoordinatorMetrics implements AutoC
public static final String CONSUMER_GROUP_REBALANCES_SENSOR_NAME =
"ConsumerGroupRebalances";
public static final String SHARE_GROUP_REBALANCES_SENSOR_NAME =
"ShareGroupRebalances";
public static final String STREAMS_GROUP_REBALANCES_SENSOR_NAME =
"StreamsGroupRebalances";
+ public static final String
STREAMS_GROUP_TOPOLOGY_DESCRIPTION_CLEANUP_CYCLE_RUNS_SENSOR_NAME =
"StreamsGroupTopologyDescriptionCleanupCycleRuns";
+ public static final String
STREAMS_GROUP_TOPOLOGY_DESCRIPTION_CLEANUP_ELIGIBLE_GROUPS_SENSOR_NAME =
"StreamsGroupTopologyDescriptionCleanupEligibleGroups";
+ public static final String
STREAMS_GROUP_TOPOLOGY_DESCRIPTION_DELETE_SUCCESS_SENSOR_NAME =
"StreamsGroupTopologyDescriptionDeleteSuccess";
+ public static final String
STREAMS_GROUP_TOPOLOGY_DESCRIPTION_DELETE_ERROR_SENSOR_NAME =
"StreamsGroupTopologyDescriptionDeleteError";
private final MetricName offsetCountMetricName;
private final MetricName classicGroupCountMetricName;
@@ -398,6 +402,46 @@ public class GroupCoordinatorMetrics extends
CoordinatorMetrics implements AutoC
METRICS_GROUP,
"The total number of streams group rebalances")));
+ Sensor streamsGroupTopologyDescriptionCleanupCycleRunsSensor =
+
metrics.sensor(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_CLEANUP_CYCLE_RUNS_SENSOR_NAME);
+ streamsGroupTopologyDescriptionCleanupCycleRunsSensor.add(new Meter(
+
metrics.metricName("streams-group-topology-description-cleanup-cycle-rate",
+ METRICS_GROUP,
+ "The rate at which the topology-description cleanup cycle
fires"),
+
metrics.metricName("streams-group-topology-description-cleanup-cycle-count",
+ METRICS_GROUP,
+ "The total number of topology-description cleanup cycles that
ran")));
+
+ Sensor streamsGroupTopologyDescriptionCleanupEligibleGroupsSensor =
+
metrics.sensor(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_CLEANUP_ELIGIBLE_GROUPS_SENSOR_NAME);
+ streamsGroupTopologyDescriptionCleanupEligibleGroupsSensor.add(new
Meter(
+
metrics.metricName("streams-group-topology-description-cleanup-eligible-rate",
+ METRICS_GROUP,
+ "The rate of streams groups identified as eligible for
topology-description cleanup"),
+
metrics.metricName("streams-group-topology-description-cleanup-eligible-count",
+ METRICS_GROUP,
+ "The total number of streams groups identified as eligible for
topology-description cleanup")));
+
+ Sensor streamsGroupTopologyDescriptionDeleteSuccessSensor =
+
metrics.sensor(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_DELETE_SUCCESS_SENSOR_NAME);
+ streamsGroupTopologyDescriptionDeleteSuccessSensor.add(new Meter(
+
metrics.metricName("streams-group-topology-description-delete-success-rate",
+ METRICS_GROUP,
+ "The rate of successful plugin.deleteTopology calls
(DeleteGroups and periodic cleanup combined)"),
+
metrics.metricName("streams-group-topology-description-delete-success-count",
+ METRICS_GROUP,
+ "The total number of successful plugin.deleteTopology calls
(DeleteGroups and periodic cleanup combined)")));
+
+ Sensor streamsGroupTopologyDescriptionDeleteErrorSensor =
+
metrics.sensor(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_DELETE_ERROR_SENSOR_NAME);
+ streamsGroupTopologyDescriptionDeleteErrorSensor.add(new Meter(
+
metrics.metricName("streams-group-topology-description-delete-error-rate",
+ METRICS_GROUP,
+ "The rate of failed plugin.deleteTopology calls (DeleteGroups
and periodic cleanup combined)"),
+
metrics.metricName("streams-group-topology-description-delete-error-count",
+ METRICS_GROUP,
+ "The total number of failed plugin.deleteTopology calls
(DeleteGroups and periodic cleanup combined)")));
+
globalSensors = Collections.unmodifiableMap(Utils.mkMap(
Utils.mkEntry(OFFSET_COMMITS_SENSOR_NAME, offsetCommitsSensor),
Utils.mkEntry(OFFSET_EXPIRED_SENSOR_NAME, offsetExpiredSensor),
@@ -405,10 +449,37 @@ public class GroupCoordinatorMetrics extends
CoordinatorMetrics implements AutoC
Utils.mkEntry(CLASSIC_GROUP_COMPLETED_REBALANCES_SENSOR_NAME,
classicGroupCompletedRebalancesSensor),
Utils.mkEntry(CONSUMER_GROUP_REBALANCES_SENSOR_NAME,
consumerGroupRebalanceSensor),
Utils.mkEntry(SHARE_GROUP_REBALANCES_SENSOR_NAME,
shareGroupRebalanceSensor),
- Utils.mkEntry(STREAMS_GROUP_REBALANCES_SENSOR_NAME,
streamsGroupRebalanceSensor)
+ Utils.mkEntry(STREAMS_GROUP_REBALANCES_SENSOR_NAME,
streamsGroupRebalanceSensor),
+
Utils.mkEntry(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_CLEANUP_CYCLE_RUNS_SENSOR_NAME,
+ streamsGroupTopologyDescriptionCleanupCycleRunsSensor),
+
Utils.mkEntry(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_CLEANUP_ELIGIBLE_GROUPS_SENSOR_NAME,
+ streamsGroupTopologyDescriptionCleanupEligibleGroupsSensor),
+
Utils.mkEntry(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_DELETE_SUCCESS_SENSOR_NAME,
+ streamsGroupTopologyDescriptionDeleteSuccessSensor),
+
Utils.mkEntry(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_DELETE_ERROR_SENSOR_NAME,
+ streamsGroupTopologyDescriptionDeleteErrorSensor)
));
}
+ /**
+ * Record a single observation against a global sensor by name. No-op if
the sensor is not
+ * configured (e.g. tests that build the metrics without the streams
plugin scaffolding).
+ */
+ public void recordSensor(String name) {
+ Sensor sensor = globalSensors.get(name);
+ if (sensor != null) sensor.record();
+ }
+
+ /**
+ * Record a numeric observation against a global sensor by name. No-op if
the sensor is
+ * not configured. Used by the topology-description cleanup cycle to
report the eligible
+ * group count per cycle.
+ */
+ public void recordSensor(String name, double value) {
+ Sensor sensor = globalSensors.get(name);
+ if (sensor != null) sensor.record(value);
+ }
+
private Long numOffsets() {
return
shards.values().stream().mapToLong(GroupCoordinatorMetricsShard::numOffsets).sum();
}
@@ -491,7 +562,11 @@ public class GroupCoordinatorMetrics extends
CoordinatorMetrics implements AutoC
CLASSIC_GROUP_COMPLETED_REBALANCES_SENSOR_NAME,
CONSUMER_GROUP_REBALANCES_SENSOR_NAME,
SHARE_GROUP_REBALANCES_SENSOR_NAME,
- STREAMS_GROUP_REBALANCES_SENSOR_NAME
+ STREAMS_GROUP_REBALANCES_SENSOR_NAME,
+ STREAMS_GROUP_TOPOLOGY_DESCRIPTION_CLEANUP_CYCLE_RUNS_SENSOR_NAME,
+
STREAMS_GROUP_TOPOLOGY_DESCRIPTION_CLEANUP_ELIGIBLE_GROUPS_SENSOR_NAME,
+ STREAMS_GROUP_TOPOLOGY_DESCRIPTION_DELETE_SUCCESS_SENSOR_NAME,
+ STREAMS_GROUP_TOPOLOGY_DESCRIPTION_DELETE_ERROR_SENSOR_NAME
).forEach(metrics::removeSensor);
}
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java
index 39caa8ddb9d..85c0d1d894f 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java
@@ -988,6 +988,18 @@ public class StreamsGroup implements Group {
return state() == StreamsGroupState.EMPTY;
}
+ /**
+ * Snapshot-aware counterpart to {@link #isEmpty()}: returns whether the
group was in the
+ * {@code EMPTY} state at {@code committedOffset}. Used by read operations
(e.g. the
+ * topology-description cleanup eligibility scan) that must observe
committed state only —
+ * a member-join write that has been committed but not yet applied to the
live state, or
+ * vice versa, would otherwise widen the scan-vs-clear race window beyond
what the runtime
+ * contract allows.
+ */
+ public boolean isEmpty(long committedOffset) {
+ return state.get(committedOffset) == StreamsGroupState.EMPTY;
+ }
+
/**
* See {@link org.apache.kafka.coordinator.group.OffsetExpirationCondition}
*
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java
index 61da755aa9e..5955155b9fa 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java
@@ -31,6 +31,8 @@ import org.apache.kafka.common.utils.internals.LogContext;
import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription;
import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescriptionPlugin;
import
org.apache.kafka.coordinator.group.api.streams.StreamsTopologyDescriptionPermanentFailureException;
+import org.apache.kafka.server.util.timer.Timer;
+import org.apache.kafka.server.util.timer.TimerTask;
import org.slf4j.Logger;
@@ -43,18 +45,22 @@ import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
import static
org.apache.kafka.common.requests.StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_AVAILABLE;
import static
org.apache.kafka.common.requests.StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_ERROR;
import static
org.apache.kafka.common.requests.StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_NOT_STORED;
/**
- * Broker-level component that owns the streams-group topology description
plugin
- * reference and the per-group re-solicitation back-off. The chain that drives
a push
- * RPC (validate → convert → plugin → metadata write → back-off mutation)
lives on
- * {@code GroupCoordinatorService}; this class exposes the building blocks
(plugin
- * invocation, back-off mutations) and the heartbeat-path gate, but does not
assemble
- * the chain itself.
+ * Broker-level component that owns the streams-group topology description
plugin reference
+ * and the per-group re-solicitation back-off. The push-RPC chain (validate →
convert →
+ * plugin → metadata write → back-off mutation) and the periodic cleanup
cycle's body live
+ * on {@code GroupCoordinatorService}; this class exposes the building blocks
+ * ({@link #invokeSetTopology}, {@link #completeEpochWrite}, {@link
#armBackoff},
+ * {@link #invokeDeleteTopologies}, {@link #clearBackoffGroup}) and one
harness method,
+ * {@link #startCleanupCycle}, that wraps a service-supplied cycle body with
single-flight
+ * scheduling on the broker timer.
*
* <p>This class is broker-level (one instance per {@code
GroupCoordinatorService}); the
* back-off map is keyed by {@code groupId} and shared across all partitions
hosted on the
@@ -68,6 +74,29 @@ public class StreamsGroupTopologyDescriptionManager
implements AutoCloseable {
private final Optional<StreamsGroupTopologyDescriptionPlugin> plugin;
private final StreamsGroupTopologyDescriptionBackoff backoff;
+ /**
+ * True between {@link #startCleanupCycle} and {@link #close}. The {@link
TimerTask}
+ * checks this on each tick before invoking the cycle supplier, so {@code
close} flips
+ * the flag and the next tick refuses to fire even if {@code cancel} on
the task races.
+ */
+ private final AtomicBoolean running = new AtomicBoolean(false);
+
+ /**
+ * Single-flight guard for the periodic cleanup cycle: a tick that fires
while the
+ * previous cycle is still settling is dropped with a warn-level log. Set
true in
+ * {@link #runOnce} before invoking the supplier; released by the terminal
+ * {@code whenComplete} attached to the future the supplier returns.
+ */
+ private final AtomicBoolean cycleInFlight = new AtomicBoolean(false);
+
+ /**
+ * The currently-scheduled cleanup tick on the broker-level {@link Timer}.
+ * Self-rescheduled inside the {@link TimerTask}'s {@code run}; {@link
#close} cancels
+ * this snapshot and the task's own re-arm check observes {@code running
== false}
+ * so the next tick does not re-schedule itself.
+ */
+ private volatile TimerTask scheduledTask;
+
public StreamsGroupTopologyDescriptionManager(
LogContext logContext,
Optional<StreamsGroupTopologyDescriptionPlugin> plugin,
@@ -79,18 +108,108 @@ public class StreamsGroupTopologyDescriptionManager
implements AutoCloseable {
}
/**
- * Release plugin-side resources. The plugin is instantiated by the
service via
- * {@code config.getConfiguredInstance(...)}, so the service owns it and
must close
- * it on shutdown to avoid leaking threads, network clients, etc. across
broker
- * restart cycles.
+ * Arm the periodic cleanup cycle. The manager owns the scheduling harness
— timer task,
+ * single-flight guard, running flag — and fires the service-supplied
{@code cycleSupplier}
+ * on every tick; the cycle body (which operations to schedule on the
runtime in what
+ * order) lives entirely on the service side. No-op when no plugin is
configured. Must
+ * be called before {@link #close}; a second call while already running
logs and is
+ * otherwise a no-op.
+ */
+ public void startCleanupCycle(
+ Timer timer,
+ long cleanupCheckIntervalMs,
+ Supplier<CompletableFuture<?>> cycleSupplier
+ ) {
+ if (plugin.isEmpty()) return;
+ if (!running.compareAndSet(false, true)) {
+ log.warn("Topology-description cleanup cycle is already started.");
+ return;
+ }
+ 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 {
+ if (running.compareAndSet(true, false)) {
+ TimerTask snapshot = scheduledTask;
+ if (snapshot != null) {
+ snapshot.cancel();
+ }
+ }
if (plugin.isPresent()) {
plugin.get().close();
}
}
+ private void scheduleNextTick(
+ Timer timer,
+ long cleanupCheckIntervalMs,
+ Supplier<CompletableFuture<?>> cycleSupplier
+ ) {
+ if (!running.get()) return;
+ TimerTask task = new TimerTask(cleanupCheckIntervalMs) {
+ @Override
+ public void run() {
+ if (!running.get()) return;
+ try {
+ runOnce(cycleSupplier);
+ } catch (Throwable t) {
+ log.warn("Unexpected error running topology-description
cleanup cycle.", t);
+ }
+ if (running.get()) scheduleNextTick(timer,
cleanupCheckIntervalMs, cycleSupplier);
+ }
+ };
+ scheduledTask = task;
+ timer.add(task);
+ }
+
+ /**
+ * Invoke {@code cycleSupplier} once under the single-flight guard: a call
that fires
+ * while a previous cycle is still settling its returned future is dropped
with a
+ * warn-level log. Released in the terminal {@code whenComplete}; a
synchronous throw
+ * from the supplier releases the flag before propagating so the next tick
can run.
+ */
+ // Visible for testing.
+ public void runOnce(Supplier<CompletableFuture<?>> cycleSupplier) {
+ if (!cycleInFlight.compareAndSet(false, true)) {
+ log.warn("Topology-description cleanup cycle skipped: previous
cycle is still in flight.");
+ return;
+ }
+ try {
+ CompletableFuture<?> chain = cycleSupplier.get();
+ if (chain == null) {
+ cycleInFlight.set(false);
+ return;
+ }
+ chain.whenComplete((__, throwable) -> {
+ if (throwable != null) {
+ log.warn("Topology-description cleanup cycle failed to
complete cleanly.", throwable);
+ }
+ cycleInFlight.set(false);
+ });
+ } catch (Throwable t) {
+ cycleInFlight.set(false);
+ throw t;
+ }
+ }
+
+ // Visible for testing.
+ TimerTask scheduledCleanupTask() {
+ return scheduledTask;
+ }
+
+ // Visible for testing.
+ boolean isRunning() {
+ return running.get();
+ }
+
/**
* @return true if a topology description plugin is configured on this
broker.
*/
@@ -263,10 +382,10 @@ public class StreamsGroupTopologyDescriptionManager
implements AutoCloseable {
* SPI contract) is mapped to the same {@code GROUP_DELETION_FAILED} as an
* exceptional future.
*
- * <p>Pure plugin invocation: does not read group state and does not touch
the
- * back-off map. The service layer pre-filters the input via
- * {@code streamsGroupsWithStoredTopologyDescription} and is responsible
for invoking
- * {@link #clearBackoffGroup} for the groups that were attempted.
+ * <p>Pure plugin invocation: does not read group state, does not touch
the back-off
+ * map, and does not record metrics. The service layer pre-filters the
input, records
+ * delete-success / delete-error sensors on the returned failure count,
and is
+ * responsible for invoking {@link #clearBackoffGroup} for the groups it
chose to clear.
*/
public CompletableFuture<Map<String, ApiError>>
invokeDeleteTopologies(Set<String> groupIds) {
if (plugin.isEmpty() || groupIds.isEmpty()) {
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java
index 43a06050339..204e1e95509 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java
@@ -47,6 +47,7 @@ import org.apache.kafka.server.util.timer.MockTimer;
import org.junit.jupiter.api.Test;
+import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -63,6 +64,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
@@ -917,6 +919,293 @@ public class
GroupCoordinatorServiceTopologyDescriptionTest {
assertEquals("Topology description plugin failed to delete the
topology.", badResult.errorMessage());
}
+ @Test
+ public void testCleanupCycleNoOpWhenNoPlugin() {
+ // No plugin configured -> the cycle must not even touch the runtime.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ GroupCoordinatorService service = buildService(runtime,
Optional.empty(), true);
+
+ service.runOneStreamsTopologyCleanupCycle();
+
+ verify(runtime,
never()).scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"),
any());
+ verify(runtime,
never()).scheduleWriteOperation(eq("clear-stored-topology-epoch"), any(),
any());
+ }
+
+ @Test
+ public void testCleanupCycleClearsStoredEpochOnPluginSuccess() {
+ // Eligibility scan returns one group at storedEpoch=4; plugin
succeeds; the cycle must
+ // schedule the conditional clear-stored write echoing the same epoch
back so a
+ // concurrent setTopology that has advanced the field is preserved.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+
when(plugin.deleteTopology("foo")).thenReturn(CompletableFuture.completedFuture(null));
+
when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"),
any()))
+
.thenReturn(List.of(CompletableFuture.completedFuture(Map.of("foo", 4))));
+ when(runtime.scheduleWriteOperation(eq("clear-stored-topology-epoch"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(null));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ service.runOneStreamsTopologyCleanupCycle();
+
+ verify(plugin, times(1)).deleteTopology("foo");
+ verify(runtime,
times(1)).scheduleWriteOperation(eq("clear-stored-topology-epoch"),
eq(GROUP_TP), any());
+ }
+
+ @Test
+ public void testCleanupCycleSkipsClearOnPluginFailure() {
+ // Plugin fails -> the cycle must NOT clear stored epoch; the group
stays gated on
+ // the next sweep and the next cycle retries the plugin call.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(plugin.deleteTopology("foo"))
+ .thenReturn(CompletableFuture.failedFuture(new
RuntimeException("plugin offline")));
+
when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"),
any()))
+
.thenReturn(List.of(CompletableFuture.completedFuture(Map.of("foo", 4))));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ service.runOneStreamsTopologyCleanupCycle();
+
+ verify(plugin, times(1)).deleteTopology("foo");
+ verify(runtime,
never()).scheduleWriteOperation(eq("clear-stored-topology-epoch"), any(),
any());
+ }
+
+ @Test
+ public void testCleanupCyclePreservesBackoffOnPluginFailure() throws
Exception {
+ // Unconditionally clearing the broker-wide back-off entry on
+ // a failed plugin.deleteTopology bypasses push-path ratchet for any
group
+ // the cycle touches. If a member rejoins between the failing scan and
the next cycle,
+ // the push-path back-off check finds no entry and re-attacks the
broken plugin at
+ // attempts=0 every join. The cycle must leave the entry in place so
the existing
+ // exponential window still throttles concurrent set-topology pushes.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(plugin.deleteTopology("foo"))
+ .thenReturn(CompletableFuture.failedFuture(new
RuntimeException("plugin offline")));
+
when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"),
any()))
+
.thenReturn(List.of(CompletableFuture.completedFuture(Map.of("foo", 4))));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ // Arm a back-off entry at the same currentEpoch we will probe with
the heartbeat helper,
+ // then run a failing cycle. The helper's gate calls armIfNotActive at
that epoch — if
+ // the cycle wiped the entry the heartbeat would arm freshly and set
the flag; if the
+ // entry survived the heartbeat sees an active window and the flag
stays unset.
+ service.streamsGroupTopologyDescriptionManager().armBackoff("foo", 4);
+ service.runOneStreamsTopologyCleanupCycle();
+
+ assertFalse(heartbeatTopologyDescriptionRequired(runtime, service, 4,
2, -1),
+ "failed cycle must not clear the back-off entry");
+ }
+
+ @Test
+ public void testCleanupCycleClearsBackoffOnPluginSuccess() throws
Exception {
+ // Symmetric counterpart: a successful plugin.deleteTopology means the
group is on its
+ // way to tombstone — the back-off entry is no longer load-bearing for
any future state
+ // of this groupId. The cycle clears it; a subsequent re-creation of
the same id is a
+ // fresh lifecycle and arms a fresh chain.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+
when(plugin.deleteTopology("foo")).thenReturn(CompletableFuture.completedFuture(null));
+
when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"),
any()))
+
.thenReturn(List.of(CompletableFuture.completedFuture(Map.of("foo", 4))));
+ when(runtime.scheduleWriteOperation(eq("clear-stored-topology-epoch"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(null));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ service.streamsGroupTopologyDescriptionManager().armBackoff("foo", 4);
+ service.runOneStreamsTopologyCleanupCycle();
+
+ assertTrue(heartbeatTopologyDescriptionRequired(runtime, service, 4,
2, -1),
+ "successful cycle must clear the back-off so a fresh solicitation
can arm");
+ }
+
+ @Test
+ public void testCleanupCycleEmptyEligibility() {
+ // No groups eligible -> plugin is not called and no clear write is
scheduled.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+
when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"),
any()))
+ .thenReturn(List.of(CompletableFuture.completedFuture(Map.of())));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ service.runOneStreamsTopologyCleanupCycle();
+
+ verify(plugin, never()).deleteTopology(anyString());
+ verify(runtime,
never()).scheduleWriteOperation(eq("clear-stored-topology-epoch"), any(),
any());
+ }
+
+ @Test
+ public void testCleanupCycleSingleFlightSkipsConcurrentCycle() {
+ // The first cycle's per-partition read is parked on an unresolved
future. A second
+ // call must observe cleanupCycleInFlight and skip without scheduling
another read.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+
when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"),
any()))
+ .thenReturn(List.of(new CompletableFuture<>()));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+
service.streamsGroupTopologyDescriptionManager().runOnce(service::runOneStreamsTopologyCleanupCycle);
+
service.streamsGroupTopologyDescriptionManager().runOnce(service::runOneStreamsTopologyCleanupCycle);
+
+ verify(runtime,
times(1)).scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"),
any());
+ }
+
+ @Test
+ public void testCleanupCycleSingleFlightHoldsFlagUntilClearWriteSettles() {
+ // Locks the fix for the gap Copilot flagged: invokeDeleteTopologies's
plugin call
+ // completes synchronously, but the conditional clear-stored-epoch
write is parked
+ // on an unresolved future. Until that write settles, the in-flight
flag must remain
+ // held — a fresh cycle scheduled by the timer would otherwise re-scan
the same
+ // eligible group (storedEpoch still != -1 because the clear has not
landed) and
+ // double-fire plugin.deleteTopology. After the parked write completes
the flag is
+ // released and a subsequent cycle observes a fresh
scheduleReadAllOperation.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+
when(plugin.deleteTopology("foo")).thenReturn(CompletableFuture.completedFuture(null));
+
when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"),
any()))
+
.thenReturn(List.of(CompletableFuture.completedFuture(Map.of("foo", 4))));
+ CompletableFuture<Object> parkedClearWrite = new CompletableFuture<>();
+ when(runtime.scheduleWriteOperation(eq("clear-stored-topology-epoch"),
eq(GROUP_TP), any()))
+ .thenReturn(parkedClearWrite);
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+
service.streamsGroupTopologyDescriptionManager().runOnce(service::runOneStreamsTopologyCleanupCycle);
+ // Plugin call resolved synchronously but clear-write is parked —
second cycle skipped.
+
service.streamsGroupTopologyDescriptionManager().runOnce(service::runOneStreamsTopologyCleanupCycle);
+ verify(runtime,
times(1)).scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"),
any());
+
+ // Settle the clear-write: flag should now release, next cycle scans
afresh.
+ parkedClearWrite.complete(null);
+
service.streamsGroupTopologyDescriptionManager().runOnce(service::runOneStreamsTopologyCleanupCycle);
+ verify(runtime,
times(2)).scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"),
any());
+ }
+
+ @Test
+ public void
testCleanupCycleSingleFlightReleasesFlagOnSynchronousThrowDuringChainConstruction()
{
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+
when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"),
any()))
+ .thenThrow(new RuntimeException("synthetic runtime failure during
chain construction"))
+ .thenReturn(List.of());
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ // First tick throws — the cycle must still release the flag so a
subsequent tick runs.
+ assertThrows(RuntimeException.class,
+ () ->
service.streamsGroupTopologyDescriptionManager().runOnce(service::runOneStreamsTopologyCleanupCycle));
+ // Second tick must reach the runtime read, confirming the flag was
released.
+
service.streamsGroupTopologyDescriptionManager().runOnce(service::runOneStreamsTopologyCleanupCycle);
+
+ verify(runtime, times(2)).scheduleReadAllOperation(
+ eq("list-streams-groups-needing-topology-cleanup"), any());
+ }
+
+ @Test
+ public void testCleanupCycleSingleFlightReleasesFlagAfterCycleCompletes() {
+ // The skip case alone does not prove the flag is ever released: a
buggy whenComplete
+ // (e.g., missing the partitionDone allOf join) would leave it set
forever and silently
+ // disable every subsequent cycle. Drive a full cycle to completion
(read resolves,
+ // plugin delete settles, conditional clear write settles), then issue
a second cycle
+ // and verify it observes the released flag by scheduling a fresh read.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+
when(plugin.deleteTopology("foo")).thenReturn(CompletableFuture.completedFuture(null));
+
when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"),
any()))
+
.thenReturn(List.of(CompletableFuture.completedFuture(Map.of("foo", 4))));
+ when(runtime.scheduleWriteOperation(eq("clear-stored-topology-epoch"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(null));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+
service.streamsGroupTopologyDescriptionManager().runOnce(service::runOneStreamsTopologyCleanupCycle);
+ // Same runtime, second invocation: must schedule another read (flag
released).
+
service.streamsGroupTopologyDescriptionManager().runOnce(service::runOneStreamsTopologyCleanupCycle);
+
+ verify(runtime,
times(2)).scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"),
any());
+ }
+
+ @Test
+ public void testCleanupCycleSkipsFollowUpWorkOncePastShutdown() throws
Exception {
+ // TimerTask.cancel() does not await an in-flight cycle, so a cycle
that has already
+ // passed the CAS when shutdown fires would otherwise run
plugin.deleteTopology and
+ // follow-up scheduleWriteOperation against a manager and runtime that
are about to
+ // close. The per-partition handle inside the manager's cycle now
checks the
+ // manager's running flag before dispatching the plugin call; this
locks that
+ // behavior.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ CompletableFuture<Map<String, Integer>> parkedRead = new
CompletableFuture<>();
+ when(runtime.<Map<String,
Integer>>scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"),
any()))
+ .thenReturn(List.of(parkedRead));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ // Cycle dispatched, parked on the per-partition read.
+ service.runOneStreamsTopologyCleanupCycle();
+ // service.shutdown() closes the manager which flips its running flag
false and
+ // cancels the scheduled task (and closes the mocked runtime; mocks
remain callable
+ // for verification).
+ service.shutdown();
+ // Now resolve the read: the handle runs under running==false and must
skip the
+ // plugin dispatch + the conditional clear writes that would have
followed.
+ parkedRead.complete(Map.of("foo", 4));
+
+ verify(plugin, never()).deleteTopology(anyString());
+ verify(runtime,
never()).scheduleWriteOperation(eq("clear-stored-topology-epoch"), any(),
any());
+ }
+
+ @Test
+ public void testShutdownCancelsScheduledCleanupTask() throws Exception {
+ // startup() with a plugin configured arms the manager's periodic
cleanup tick;
+ // shutdown() must close the manager so the timer queue does not
retain a
+ // self-rescheduling task referencing a torn-down runtime.
MockTimer.size() filters
+ // cancelled entries, so observing 1 → 0 confirms manager.close()'s
cancel() landed;
+ // advancing the clock past the interval afterwards must not fire the
task — both
+ // the queue-skip on cancellation and the TimerTask body's
running==false defensive
+ // return guard against it.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ MockTimer timer = new MockTimer();
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true, timer);
+
+ assertEquals(1, timer.size());
+
+ service.shutdown();
+
+ assertEquals(0, timer.size());
+ // No cycle should fire even with the clock advanced well past the
cleanup interval.
+ timer.advanceClock(Duration.ofHours(2).toMillis());
+ verify(runtime, never()).scheduleReadAllOperation(
+ eq("list-streams-groups-needing-topology-cleanup"), any());
+ }
+
+ @Test
+ public void testShutdownSafeWhenNoCleanupTaskScheduled() {
+ // Plugin absent => manager.startCleanupCycle short-circuits before
scheduling a tick,
+ // so the scheduledTask field stays null. shutdown() must tolerate the
null snapshot
+ // without throwing — broker close paths must not propagate.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ MockTimer timer = new MockTimer();
+ GroupCoordinatorService service = buildService(runtime,
Optional.empty(), true, timer);
+
+ assertEquals(0, timer.size());
+ service.shutdown();
+ assertEquals(0, timer.size());
+ }
+
+ @Test
+ public void testCleanupCycleSingleFlightReleasesFlagOnEmptyPartitionList()
{
+ // Pathological boundary: zero hosted partitions (e.g. broker just
started, nothing
+ // loaded yet). partitionFutures is empty -> allOf(empty) -> immediate
completion ->
+ // whenComplete still fires and releases the flag. A subsequent cycle
must run.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+
when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"),
any()))
+ .thenReturn(List.of());
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+
service.streamsGroupTopologyDescriptionManager().runOnce(service::runOneStreamsTopologyCleanupCycle);
+
service.streamsGroupTopologyDescriptionManager().runOnce(service::runOneStreamsTopologyCleanupCycle);
+
+ verify(runtime,
times(2)).scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"),
any());
+ }
+
@Test
public void testDescribeWithIncludeFlagDisabledLeavesStatusDefault()
throws Exception {
// includeTopologyDescription=false -> plugin is not consulted
regardless of whether
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java
index 5f375d67c2e..ca005188ef9 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java
@@ -92,6 +92,7 @@ import
org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyKey;
import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
import org.apache.kafka.coordinator.group.modern.consumer.ConsumerGroup;
import org.apache.kafka.coordinator.group.modern.share.ShareGroup;
+import org.apache.kafka.coordinator.group.streams.StreamsGroup;
import org.apache.kafka.coordinator.group.streams.StreamsGroupDescribeResult;
import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
import org.apache.kafka.coordinator.group.streams.StreamsGroupMember;
@@ -126,16 +127,19 @@ import static
org.apache.kafka.coordinator.group.GroupCoordinatorShard.GROUP_SIZ
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -1514,6 +1518,274 @@ public class GroupCoordinatorShardTest {
verify(groupMetadataManager,
times(0)).maybeDeleteGroup(eq("other-group-id"), any());
}
+ @Test
+ public void
testListStreamsGroupsNeedingTopologyCleanupFiltersByAllPredicates() {
+ // Covers the full filter chain inside the shard's eligibility scan:
+ // - maybeGroup == null -> skip
+ // - type() != STREAMS -> skip
+ // - !isEmpty() -> skip (live members still on the group)
+ // - storedEpoch == -1 -> skip (no plugin state to clean)
+ // - !allOffsetsExpired -> skip (offsets still in retention)
+ // - empty + stored != -1 + all expired -> include, value = observed
storedEpoch
+ GroupMetadataManager groupMetadataManager =
mock(GroupMetadataManager.class);
+ OffsetMetadataManager offsetMetadataManager =
mock(OffsetMetadataManager.class);
+ GroupCoordinatorConfig config = mock(GroupCoordinatorConfig.class);
+ when(config.offsetsRetentionCheckIntervalMs()).thenReturn(60 * 60 *
1000L);
+ MockTime mockTime = new MockTime();
+ MockCoordinatorTimer<CoordinatorRecord> timer = new
MockCoordinatorTimer<>(mockTime);
+ GroupCoordinatorShard coordinator = new GroupCoordinatorShard(
+ new LogContext(),
+ groupMetadataManager,
+ offsetMetadataManager,
+ mockTime,
+ timer,
+ config,
+ mock(CoordinatorMetrics.class),
+ mock(CoordinatorMetricsShard.class)
+ );
+
+ long committedOffset = 100L;
+ // Six candidate ids, one for each filter branch.
+ when(groupMetadataManager.groupIds(committedOffset)).thenReturn(Set.of(
+ "missing", "not-streams", "non-empty", "default-stored",
"unexpired-offsets", "eligible"));
+
+ // missing: maybeGroup returns null
+ when(groupMetadataManager.maybeGroup("missing",
committedOffset)).thenReturn(null);
+
+ // not-streams: type returns CONSUMER (anything != STREAMS)
+ Group notStreams = mock(Group.class);
+ when(notStreams.type()).thenReturn(Group.GroupType.CONSUMER);
+ when(groupMetadataManager.maybeGroup("not-streams",
committedOffset)).thenReturn(notStreams);
+
+ // non-empty: STREAMS but isEmpty == false
+ StreamsGroup nonEmpty = mock(StreamsGroup.class);
+ when(nonEmpty.type()).thenReturn(Group.GroupType.STREAMS);
+ when(nonEmpty.isEmpty(committedOffset)).thenReturn(false);
+ when(groupMetadataManager.maybeGroup("non-empty",
committedOffset)).thenReturn(nonEmpty);
+
+ // default-stored: STREAMS, empty, but storedEpoch == -1
+ StreamsGroup defaultStored = mock(StreamsGroup.class);
+ when(defaultStored.type()).thenReturn(Group.GroupType.STREAMS);
+ when(defaultStored.isEmpty(committedOffset)).thenReturn(true);
+
when(defaultStored.storedDescriptionTopologyEpoch(committedOffset)).thenReturn(-1);
+ when(groupMetadataManager.maybeGroup("default-stored",
committedOffset)).thenReturn(defaultStored);
+
+ // unexpired-offsets: STREAMS, empty, storedEpoch=5, but offsets are
not all expired
+ StreamsGroup unexpired = mock(StreamsGroup.class);
+ when(unexpired.type()).thenReturn(Group.GroupType.STREAMS);
+ when(unexpired.isEmpty(committedOffset)).thenReturn(true);
+
when(unexpired.storedDescriptionTopologyEpoch(committedOffset)).thenReturn(5);
+ when(groupMetadataManager.maybeGroup("unexpired-offsets",
committedOffset)).thenReturn(unexpired);
+ when(offsetMetadataManager.allOffsetsExpired(eq("unexpired-offsets"),
anyLong(), anyLong())).thenReturn(false);
+
+ // eligible: passes every predicate; storedEpoch=7 must appear in the
result map.
+ StreamsGroup eligible = mock(StreamsGroup.class);
+ when(eligible.type()).thenReturn(Group.GroupType.STREAMS);
+ when(eligible.isEmpty(committedOffset)).thenReturn(true);
+
when(eligible.storedDescriptionTopologyEpoch(committedOffset)).thenReturn(7);
+ when(groupMetadataManager.maybeGroup("eligible",
committedOffset)).thenReturn(eligible);
+ when(offsetMetadataManager.allOffsetsExpired(eq("eligible"),
anyLong(), anyLong())).thenReturn(true);
+
+ Map<String, Integer> result =
coordinator.listStreamsGroupsNeedingTopologyCleanup(committedOffset);
+
+ assertEquals(Map.of("eligible", 7), result);
+ }
+
+ @Test
+ public void
testListStreamsGroupsNeedingTopologyCleanupSwallowsPerGroupError() {
+ // A per-group failure (e.g. unexpected ClassCastException, NPE in a
mocked path) must
+ // not abort the whole scan: the cycle is periodic and an isolated bad
group should be
+ // logged and skipped, leaving the other eligible groups in the result.
+ GroupMetadataManager groupMetadataManager =
mock(GroupMetadataManager.class);
+ OffsetMetadataManager offsetMetadataManager =
mock(OffsetMetadataManager.class);
+ GroupCoordinatorConfig config = mock(GroupCoordinatorConfig.class);
+ when(config.offsetsRetentionCheckIntervalMs()).thenReturn(60 * 60 *
1000L);
+ MockTime mockTime = new MockTime();
+ MockCoordinatorTimer<CoordinatorRecord> timer = new
MockCoordinatorTimer<>(mockTime);
+ GroupCoordinatorShard coordinator = new GroupCoordinatorShard(
+ new LogContext(),
+ groupMetadataManager,
+ offsetMetadataManager,
+ mockTime,
+ timer,
+ config,
+ mock(CoordinatorMetrics.class),
+ mock(CoordinatorMetricsShard.class)
+ );
+
+ long committedOffset = 0L;
+
when(groupMetadataManager.groupIds(committedOffset)).thenReturn(Set.of("bad",
"good"));
+
+ // bad: maybeGroup throws an unexpected RuntimeException mid-scan.
+ when(groupMetadataManager.maybeGroup("bad", committedOffset))
+ .thenThrow(new RuntimeException("synthetic scan failure"));
+
+ // good: a fully eligible group.
+ StreamsGroup good = mock(StreamsGroup.class);
+ when(good.type()).thenReturn(Group.GroupType.STREAMS);
+ when(good.isEmpty(committedOffset)).thenReturn(true);
+
when(good.storedDescriptionTopologyEpoch(committedOffset)).thenReturn(3);
+ when(groupMetadataManager.maybeGroup("good",
committedOffset)).thenReturn(good);
+ when(offsetMetadataManager.allOffsetsExpired(eq("good"), anyLong(),
anyLong())).thenReturn(true);
+
+ Map<String, Integer> result =
coordinator.listStreamsGroupsNeedingTopologyCleanup(committedOffset);
+
+ assertEquals(Map.of("good", 3), result);
+ }
+
+ @Test
+ public void
testClearStoredDescriptionTopologyEpochDelegatesToGroupMetadataManager() {
+ // The shard method is a pure delegation, but the cycle's correctness
depends on the
+ // exact (groupId, expectedStoredEpoch) tuple flowing through
unchanged and the GMM's
+ // CoordinatorResult being returned verbatim — the per-group
conditional write reads
+ // its records to decide whether the clear actually persisted.
+ GroupMetadataManager groupMetadataManager =
mock(GroupMetadataManager.class);
+ OffsetMetadataManager offsetMetadataManager =
mock(OffsetMetadataManager.class);
+ GroupCoordinatorConfig config = mock(GroupCoordinatorConfig.class);
+ when(config.offsetsRetentionCheckIntervalMs()).thenReturn(60 * 60 *
1000L);
+ MockTime mockTime = new MockTime();
+ MockCoordinatorTimer<CoordinatorRecord> timer = new
MockCoordinatorTimer<>(mockTime);
+ GroupCoordinatorShard coordinator = new GroupCoordinatorShard(
+ new LogContext(),
+ groupMetadataManager,
+ offsetMetadataManager,
+ mockTime,
+ timer,
+ config,
+ mock(CoordinatorMetrics.class),
+ mock(CoordinatorMetricsShard.class)
+ );
+
+ CoordinatorResult<Void, CoordinatorRecord> expected = new
CoordinatorResult<>(List.of());
+
when(groupMetadataManager.clearStoredDescriptionTopologyEpoch("group-id",
7)).thenReturn(expected);
+
+ assertSame(expected,
coordinator.clearStoredDescriptionTopologyEpoch("group-id", 7));
+
verify(groupMetadataManager).clearStoredDescriptionTopologyEpoch("group-id", 7);
+ }
+
+ @Test
+ public void
testCleanupGroupMetadataDefersStreamsGroupWithStoredTopologyEpoch() {
+ // Plugin configured + streams group + storedEpoch != -1 + all offsets
expired:
+ // the gate must skip maybeDeleteGroup so the broker-level
topology-cleanup cycle
+ // can drive plugin.deleteTopology and clear the field first. The
offset tombstones
+ // are still written — only the group tombstone is deferred.
+ GroupMetadataManager groupMetadataManager =
mock(GroupMetadataManager.class);
+ OffsetMetadataManager offsetMetadataManager =
mock(OffsetMetadataManager.class);
+ GroupCoordinatorConfig config = mock(GroupCoordinatorConfig.class);
+
when(config.isStreamsGroupTopologyDescriptionPluginConfigured()).thenReturn(true);
+ when(config.offsetsRetentionCheckIntervalMs()).thenReturn(60 * 60 *
1000L);
+ Time mockTime = new MockTime();
+ MockCoordinatorTimer<CoordinatorRecord> timer = new
MockCoordinatorTimer<>(mockTime);
+ GroupCoordinatorShard coordinator = new GroupCoordinatorShard(
+ new LogContext(),
+ groupMetadataManager,
+ offsetMetadataManager,
+ mockTime,
+ timer,
+ config,
+ mock(CoordinatorMetrics.class),
+ mock(CoordinatorMetricsShard.class)
+ );
+
+ CoordinatorRecord offsetCommitTombstone =
GroupCoordinatorRecordHelpers.newOffsetCommitTombstoneRecord("group-id",
"topic", 0);
+
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<List<CoordinatorRecord>> recordsCapture =
ArgumentCaptor.forClass(List.class);
+
+ StreamsGroup streamsGroup = mock(StreamsGroup.class);
+ when(streamsGroup.shouldExpire()).thenReturn(true);
+ when(streamsGroup.type()).thenReturn(Group.GroupType.STREAMS);
+ when(streamsGroup.storedDescriptionTopologyEpoch()).thenReturn(4);
+
+ when(groupMetadataManager.groupIds()).thenReturn(Set.of("group-id"));
+ when(groupMetadataManager.group("group-id")).thenReturn(streamsGroup);
+ when(offsetMetadataManager.cleanupExpiredOffsets(eq("group-id"),
recordsCapture.capture()))
+ .thenAnswer(invocation -> {
+ recordsCapture.getValue().add(offsetCommitTombstone);
+ return true;
+ });
+
+ CoordinatorResult<Void, CoordinatorRecord> result =
coordinator.cleanupGroupMetadata();
+
+ // Offset tombstone went through; group tombstone is deferred.
+ assertEquals(List.of(offsetCommitTombstone), result.records());
+ verify(offsetMetadataManager,
times(1)).cleanupExpiredOffsets(eq("group-id"), any());
+ verify(groupMetadataManager, never()).maybeDeleteGroup(eq("group-id"),
any());
+ }
+
+ @Test
+ public void
testCleanupGroupMetadataTombstoneStreamsGroupWithoutStoredTopologyEpoch() {
+ // Plugin configured + streams group + storedEpoch == -1: the gate
must NOT fire, so
+ // the group is tombstoned via the normal path. This is the
steady-state after the
+ // topology-cleanup cycle has cleared the field on the previous tick.
+ GroupMetadataManager groupMetadataManager =
mock(GroupMetadataManager.class);
+ OffsetMetadataManager offsetMetadataManager =
mock(OffsetMetadataManager.class);
+ GroupCoordinatorConfig config = mock(GroupCoordinatorConfig.class);
+
when(config.isStreamsGroupTopologyDescriptionPluginConfigured()).thenReturn(true);
+ when(config.offsetsRetentionCheckIntervalMs()).thenReturn(60 * 60 *
1000L);
+ Time mockTime = new MockTime();
+ MockCoordinatorTimer<CoordinatorRecord> timer = new
MockCoordinatorTimer<>(mockTime);
+ GroupCoordinatorShard coordinator = new GroupCoordinatorShard(
+ new LogContext(),
+ groupMetadataManager,
+ offsetMetadataManager,
+ mockTime,
+ timer,
+ config,
+ mock(CoordinatorMetrics.class),
+ mock(CoordinatorMetricsShard.class)
+ );
+
+ StreamsGroup streamsGroup = mock(StreamsGroup.class);
+ when(streamsGroup.shouldExpire()).thenReturn(true);
+ when(streamsGroup.type()).thenReturn(Group.GroupType.STREAMS);
+ when(streamsGroup.storedDescriptionTopologyEpoch()).thenReturn(-1);
+
+ when(groupMetadataManager.groupIds()).thenReturn(Set.of("group-id"));
+ when(groupMetadataManager.group("group-id")).thenReturn(streamsGroup);
+ when(offsetMetadataManager.cleanupExpiredOffsets(eq("group-id"),
any())).thenReturn(true);
+
+ coordinator.cleanupGroupMetadata();
+
+ verify(groupMetadataManager,
times(1)).maybeDeleteGroup(eq("group-id"), any());
+ }
+
+ @Test
+ public void
testCleanupGroupMetadataIgnoresStoredTopologyEpochWhenNoPluginConfigured() {
+ // Plugin absent on this broker (operator unset / never set): even a
streams group with
+ // a non-default storedEpoch must NOT be deferred — no cleanup cycle
is running to
+ // clear it, and leaving the gate engaged would prevent natural
expiration forever.
+ GroupMetadataManager groupMetadataManager =
mock(GroupMetadataManager.class);
+ OffsetMetadataManager offsetMetadataManager =
mock(OffsetMetadataManager.class);
+ GroupCoordinatorConfig config = mock(GroupCoordinatorConfig.class);
+
when(config.isStreamsGroupTopologyDescriptionPluginConfigured()).thenReturn(false);
+ when(config.offsetsRetentionCheckIntervalMs()).thenReturn(60 * 60 *
1000L);
+ Time mockTime = new MockTime();
+ MockCoordinatorTimer<CoordinatorRecord> timer = new
MockCoordinatorTimer<>(mockTime);
+ GroupCoordinatorShard coordinator = new GroupCoordinatorShard(
+ new LogContext(),
+ groupMetadataManager,
+ offsetMetadataManager,
+ mockTime,
+ timer,
+ config,
+ mock(CoordinatorMetrics.class),
+ mock(CoordinatorMetricsShard.class)
+ );
+
+ StreamsGroup streamsGroup = mock(StreamsGroup.class);
+ when(streamsGroup.shouldExpire()).thenReturn(true);
+ when(streamsGroup.type()).thenReturn(Group.GroupType.STREAMS);
+ // storedDescriptionTopologyEpoch is irrelevant — the gate must not
even look at it.
+ when(groupMetadataManager.groupIds()).thenReturn(Set.of("group-id"));
+ when(groupMetadataManager.group("group-id")).thenReturn(streamsGroup);
+ when(offsetMetadataManager.cleanupExpiredOffsets(eq("group-id"),
any())).thenReturn(true);
+
+ coordinator.cleanupGroupMetadata();
+
+ verify(groupMetadataManager,
times(1)).maybeDeleteGroup(eq("group-id"), any());
+ }
+
@Test
public void testCleanupGroupMetadataForShareGroup() {
GroupMetadataManager groupMetadataManager =
mock(GroupMetadataManager.class);
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
index 7acd1a4e016..fbe513e7fd3 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
@@ -10912,6 +10912,55 @@ public class GroupMetadataManagerTest {
assertEquals(describedGroup, actual.get(0));
}
+ @Test
+ public void
testClearStoredDescriptionTopologyEpochClearsWhenEpochMatches() {
+ // the cleanup cycle echoes the storedEpoch it observed at scan time
into
+ // the conditional clear. When the persisted value still matches, the
write emits a
+ // metadata record setting StoredDescriptionTopologyEpoch back to -1
while preserving
+ // the other tagged fields (FailedDescriptionTopologyEpoch in
particular).
+ String groupId = "streams-group";
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder().build();
+
context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord(
+ groupId, 1, 0L, -1, Map.of(), 7, 3));
+
+ CoordinatorResult<Void, CoordinatorRecord> result =
+
context.groupMetadataManager.clearStoredDescriptionTopologyEpoch(groupId, 7);
+
+ assertEquals(1, result.records().size());
+ // Apply the record and verify storedEpoch is cleared and failedEpoch
preserved.
+ context.replay(result.records().get(0));
+ StreamsGroup group =
context.groupMetadataManager.getStreamsGroupOrThrow(groupId);
+ assertEquals(-1, group.storedDescriptionTopologyEpoch());
+ assertEquals(3, group.failedDescriptionTopologyEpoch());
+ }
+
+ @Test
+ public void
testClearStoredDescriptionTopologyEpochNoOpsWhenEpochMismatches() {
+ // A concurrent setTopology has advanced storedEpoch between the
cycle's scan and this
+ // write. The clear must be a no-op to preserve the newer push instead
of silently
+ // undoing it.
+ String groupId = "streams-group";
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder().build();
+
context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord(
+ groupId, 1, 0L, -1, Map.of(), 9, -1));
+
+ CoordinatorResult<Void, CoordinatorRecord> result =
+
context.groupMetadataManager.clearStoredDescriptionTopologyEpoch(groupId, 7);
+
+ assertEquals(List.of(), result.records());
+ }
+
+ @Test
+ public void testClearStoredDescriptionTopologyEpochNoOpsForMissingGroup() {
+ // Missing groups must not throw — the next cycle will simply not see
them again.
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder().build();
+
+ CoordinatorResult<Void, CoordinatorRecord> result =
+
context.groupMetadataManager.clearStoredDescriptionTopologyEpoch("missing-group",
7);
+
+ assertEquals(List.of(), result.records());
+ }
+
@Test
public void
testStreamsGroupMetadataReplayRoundTripsTopologyDescriptionEpochs() {
// KIP-1331: replay must read storedDescriptionTopologyEpoch and
failedDescriptionTopologyEpoch from the record
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java
index 484a18d9393..960224f7156 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java
@@ -104,6 +104,8 @@ import static
org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -338,6 +340,13 @@ public class OffsetMetadataManagerTest {
return isOffsetsEmptyForGroup;
}
+ public boolean allOffsetsExpired(String groupId, long
currentTimestampMs) {
+ // Existing branch tests don't drive snapshot semantics — they
replay records
+ // directly, so reading at Long.MAX_VALUE is equivalent to
"latest" and exercises
+ // the same predicates the snapshot-aware path runs.
+ return offsetMetadataManager.allOffsetsExpired(groupId,
currentTimestampMs, Long.MAX_VALUE);
+ }
+
public List<OffsetFetchResponseData.OffsetFetchResponseTopics>
fetchOffsets(
String groupId,
List<OffsetFetchRequestData.OffsetFetchRequestTopics> topics,
@@ -3249,6 +3258,149 @@ public class OffsetMetadataManagerTest {
assertEquals(List.of(), records);
}
+ @Test
+ public void
testAllOffsetsExpiredReturnsTrueWhenGroupHasNoOffsetsAndNoOpenTxn() {
+ // offsetsByGroup == null branch: no offsets committed for this group,
and no pending
+ // transactional offsets either. The group's metadata is fully
expirable.
+ OffsetMetadataManagerTestContext context = new
OffsetMetadataManagerTestContext.Builder().build();
+ assertTrue(context.allOffsetsExpired("unknown-group-id",
context.time.milliseconds()));
+ }
+
+ @Test
+ public void
testAllOffsetsExpiredReturnsFalseWhenGroupHasNoOffsetsButHasOpenTxn() {
+ // offsetsByGroup == null branch with an open transaction recorded for
the group: still
+ // not expirable, the txn could land more offsets.
cleanupExpiredOffsets uses the same
+ // gate; allOffsetsExpired must agree.
+ OffsetMetadataManagerTestContext context = new
OffsetMetadataManagerTestContext.Builder()
+ .withOffsetsRetentionMinutes(1)
+ .build();
+
+ // A pending transactional commit creates an openTransactions entry
without populating
+ // the durable offsets map.
+ context.commitOffset(42L, "group-id", "foo", 0, 100L, 0,
context.time.milliseconds());
+ assertFalse(context.allOffsetsExpired("group-id",
context.time.milliseconds()));
+ }
+
+ @Test
+ public void
testAllOffsetsExpiredReturnsFalseWhenExpirationConditionEmpty() {
+ // offsetExpirationCondition.isEmpty() branch: e.g., a classic-style
group with no
+ // expiration policy. The eligibility check must conservatively return
false rather
+ // than treating the absence of a policy as "always expirable".
+ GroupMetadataManager groupMetadataManager =
mock(GroupMetadataManager.class);
+ Group group = mock(Group.class);
+ OffsetMetadataManagerTestContext context = new
OffsetMetadataManagerTestContext.Builder()
+ .withGroupMetadataManager(groupMetadataManager)
+ .build();
+
+ context.commitOffset("group-id", "foo", 0, 100L, 0);
+ when(groupMetadataManager.group(eq("group-id"),
anyLong())).thenReturn(group);
+ when(group.offsetExpirationCondition()).thenReturn(Optional.empty());
+
+ assertFalse(context.allOffsetsExpired("group-id",
context.time.milliseconds()));
+ }
+
+ @Test
+ public void testAllOffsetsExpiredReturnsFalseWhenSubscribedToTopic() {
+ // isSubscribedToTopic branch: an offset whose topic is in the group's
live
+ // subscription is not eligible for expiration regardless of age —
even after the
+ // retention window elapses, an active subscription holds the offset.
+ GroupMetadataManager groupMetadataManager =
mock(GroupMetadataManager.class);
+ Group group = mock(Group.class);
+ OffsetMetadataManagerTestContext context = new
OffsetMetadataManagerTestContext.Builder()
+ .withGroupMetadataManager(groupMetadataManager)
+ .withOffsetsRetentionMinutes(1)
+ .build();
+
+ context.commitOffset("group-id", "foo", 0, 100L, 0,
context.time.milliseconds());
+ context.time.sleep(Duration.ofMinutes(1).toMillis());
+
+ when(groupMetadataManager.group(eq("group-id"),
anyLong())).thenReturn(group);
+ when(group.offsetExpirationCondition()).thenReturn(Optional.of(
+ new OffsetExpirationConditionImpl(offsetAndMetadata ->
offsetAndMetadata.commitTimestampMs)));
+ when(group.isSubscribedToTopic("foo")).thenReturn(true);
+
+ assertFalse(context.allOffsetsExpired("group-id",
context.time.milliseconds()));
+ }
+
+ @Test
+ public void
testAllOffsetsExpiredReturnsFalseWhenPendingTransactionalOffset() {
+ // Pending transactional offset branch: an unsubscribed topic whose
retention window
+ // has elapsed is still not eligible if a transactional offset is
pending on the same
+ // partition — the txn could commit a fresh value the cleanup would
lose.
+ GroupMetadataManager groupMetadataManager =
mock(GroupMetadataManager.class);
+ Group group = mock(Group.class);
+ OffsetMetadataManagerTestContext context = new
OffsetMetadataManagerTestContext.Builder()
+ .withGroupMetadataManager(groupMetadataManager)
+ .withOffsetsRetentionMinutes(1)
+ .build();
+
+ long commitTimestamp = context.time.milliseconds();
+ context.commitOffset("group-id", "foo", 0, 100L, 0, commitTimestamp);
+ // Concurrent transactional commit on the same partition; not yet
visible.
+ context.commitOffset(10L, "group-id", "foo", 0, 101L, 0,
commitTimestamp + 500);
+ context.time.sleep(Duration.ofMinutes(1).toMillis());
+
+ when(groupMetadataManager.group(eq("group-id"),
anyLong())).thenReturn(group);
+ when(group.offsetExpirationCondition()).thenReturn(Optional.of(
+ new OffsetExpirationConditionImpl(offsetAndMetadata ->
offsetAndMetadata.commitTimestampMs)));
+ when(group.isSubscribedToTopic("foo")).thenReturn(false);
+
+ assertFalse(context.allOffsetsExpired("group-id",
context.time.milliseconds()));
+ }
+
+ @Test
+ public void
testAllOffsetsExpiredReturnsFalseWhenOpenTxnOnPartitionWithoutCommittedOffset()
{
+ // Trailing group-level !openTransactions.contains check when
offsetsByTopic is non-null:
+ // foo/0 has an expired committed offset and no pending txn on the
same partition (the
+ // per-partition loop sees it as expirable), but a transactional
commit is in flight on
+ // bar/0 with no durable committed offset there. The per-partition
loop never visits
+ // bar/0 because offsetsByGroup has no entry for bar, so only the
trailing group-level
+ // check catches the open transaction and keeps the group ineligible.
+ GroupMetadataManager groupMetadataManager =
mock(GroupMetadataManager.class);
+ Group group = mock(Group.class);
+ OffsetMetadataManagerTestContext context = new
OffsetMetadataManagerTestContext.Builder()
+ .withGroupMetadataManager(groupMetadataManager)
+ .withOffsetsRetentionMinutes(1)
+ .build();
+
+ long commitTimestamp = context.time.milliseconds();
+ context.commitOffset("group-id", "foo", 0, 100L, 0, commitTimestamp);
+ // Transactional commit on a different topic-partition; never visited
by the per-partition
+ // loop in allOffsetsExpired because offsetsByGroup has no entry for
bar.
+ context.commitOffset(10L, "group-id", "bar", 0, 200L, 0,
commitTimestamp + 500);
+ context.time.sleep(Duration.ofMinutes(1).toMillis());
+
+ when(groupMetadataManager.group(eq("group-id"),
anyLong())).thenReturn(group);
+ when(group.offsetExpirationCondition()).thenReturn(Optional.of(
+ new OffsetExpirationConditionImpl(offsetAndMetadata ->
offsetAndMetadata.commitTimestampMs)));
+ when(group.isSubscribedToTopic("foo")).thenReturn(false);
+
+ assertFalse(context.allOffsetsExpired("group-id",
context.time.milliseconds()));
+ }
+
+ @Test
+ public void
testAllOffsetsExpiredReturnsTrueWhenAllOffsetsPastRetentionAndUnsubscribed() {
+ // Happy path: an unsubscribed topic whose offset has aged past
retention, no pending
+ // transactional offsets. The group is fully eligible for the
downstream cleanup pass.
+ GroupMetadataManager groupMetadataManager =
mock(GroupMetadataManager.class);
+ Group group = mock(Group.class);
+ OffsetMetadataManagerTestContext context = new
OffsetMetadataManagerTestContext.Builder()
+ .withGroupMetadataManager(groupMetadataManager)
+ .withOffsetsRetentionMinutes(1)
+ .build();
+
+ long commitTimestamp = context.time.milliseconds();
+ context.commitOffset("group-id", "foo", 0, 100L, 0, commitTimestamp);
+ context.time.sleep(Duration.ofMinutes(1).toMillis());
+
+ when(groupMetadataManager.group(eq("group-id"),
anyLong())).thenReturn(group);
+ when(group.offsetExpirationCondition()).thenReturn(Optional.of(
+ new OffsetExpirationConditionImpl(offsetAndMetadata ->
offsetAndMetadata.commitTimestampMs)));
+ when(group.isSubscribedToTopic("foo")).thenReturn(false);
+
+ assertTrue(context.allOffsetsExpired("group-id",
context.time.milliseconds()));
+ }
+
private static OffsetFetchResponseData.OffsetFetchResponsePartitions
mkOffsetPartitionResponse(
int partition,
long offset,
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsTest.java
index 70da4657464..df031f07935 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsTest.java
@@ -150,6 +150,14 @@ public class GroupCoordinatorMetricsTest {
Map.of("protocol", Group.GroupType.STREAMS.toString())),
metrics.metricName("streams-group-rebalance-rate",
GroupCoordinatorMetrics.METRICS_GROUP),
metrics.metricName("streams-group-rebalance-count",
GroupCoordinatorMetrics.METRICS_GROUP),
+
metrics.metricName("streams-group-topology-description-cleanup-cycle-rate",
GroupCoordinatorMetrics.METRICS_GROUP),
+
metrics.metricName("streams-group-topology-description-cleanup-cycle-count",
GroupCoordinatorMetrics.METRICS_GROUP),
+
metrics.metricName("streams-group-topology-description-cleanup-eligible-rate",
GroupCoordinatorMetrics.METRICS_GROUP),
+
metrics.metricName("streams-group-topology-description-cleanup-eligible-count",
GroupCoordinatorMetrics.METRICS_GROUP),
+
metrics.metricName("streams-group-topology-description-delete-success-rate",
GroupCoordinatorMetrics.METRICS_GROUP),
+
metrics.metricName("streams-group-topology-description-delete-success-count",
GroupCoordinatorMetrics.METRICS_GROUP),
+
metrics.metricName("streams-group-topology-description-delete-error-rate",
GroupCoordinatorMetrics.METRICS_GROUP),
+
metrics.metricName("streams-group-topology-description-delete-error-count",
GroupCoordinatorMetrics.METRICS_GROUP),
metrics.metricName(
"streams-group-count",
GroupCoordinatorMetrics.METRICS_GROUP,
