This is an automated email from the ASF dual-hosted git repository.
frankvicky 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 32d24f93a36 MINOR: Follow-up KAFKA-20696 (#22676)
32d24f93a36 is described below
commit 32d24f93a36ed2a154495a3395f1f4bb2c75c50d
Author: TengYao Chi <[email protected]>
AuthorDate: Tue Jun 30 13:49:28 2026 +0100
MINOR: Follow-up KAFKA-20696 (#22676)
Deal the comments of #22622
- **`OffsetMetadataManager.allOffsetsExpired` → `groupHasNoOffsets`**.
Per-topic / per-partition expiration check was O(topics × partitions)
per group. Since the regular offset-expiration cycle already tombstones
expired offsets, this stage only needs the post-state — a single
null-check on `offsetsByGroup` plus an `openTransactions.contains`
(O(1)). Three cycles compose cleanly: offset expiration → topology
cleanup → group tombstone. The now-unused snapshot overload
`OpenTransactions.contains(groupId, topic, partition, committedOffset)`
is removed.
- **Per-shard batching of the conditional clear write**. All groups in
one partition's eligibility read hash to the same `__consumer_offsets`
partition, so the cycle now issues one `scheduleWriteOperation` per
shard carrying every conditional clear for that shard — not one write
per group. Added
`GroupCoordinatorShard.clearStoredDescriptionTopologyEpochBatch(Map<String,
Integer>)` that folds per-group GMM results into one
`CoordinatorResult`.
- **Restored inline comment** on the pending-transactional-offset guard
in `cleanupExpiredOffsets` — accidentally dropped during the
helper-extract / inline round-trip.
Reviewers: Alieh Saeedi <[email protected]>, Lucas Brutschy
<[email protected]>
---
.../org/apache/kafka/coordinator/group/Group.java | 17 ++-
.../coordinator/group/GroupCoordinatorService.java | 40 ++++--
.../coordinator/group/GroupCoordinatorShard.java | 69 +++++-----
.../coordinator/group/OffsetMetadataManager.java | 75 ++--------
.../coordinator/group/modern/share/ShareGroup.java | 5 +-
.../coordinator/group/streams/StreamsGroup.java | 19 +++
...pCoordinatorServiceTopologyDescriptionTest.java | 27 ++++
.../group/GroupCoordinatorShardTest.java | 127 +++++++++--------
.../group/OffsetMetadataManagerTest.java | 153 +++------------------
.../group/streams/StreamsGroupTest.java | 40 ++++++
10 files changed, 268 insertions(+), 304 deletions(-)
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java
index e0bff1d51c4..a656c8918ee 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java
@@ -210,11 +210,22 @@ public interface Group {
void requestMetadataRefresh();
/**
- * Returns whether this group should be expired or not.
+ * Whether this group's metadata record should be tombstoned by the
natural-expiration sweep
+ * after its offsets have been expired. Returning {@code false} preserves
the group's
+ * metadata for the current sweep — useful when a group-type-specific
async cleanup needs to
+ * run first (e.g. the streams topology-description plugin cycle). The
next sweep
+ * re-evaluates and proceeds once the override returns {@code true}.
*
- * @return whether the group should be expired.
+ * <p><b>This does not prevent expiration of the group's offsets.</b> The
caller always runs
+ * {@code cleanupExpiredOffsets} for the group before consulting this
method, so a
+ * {@code false} return only suppresses the group-metadata tombstone —
committed offsets
+ * past {@code offsets.retention.ms} are still removed.
+ *
+ * @param config The broker-level group coordinator config; passed so
per-group-type overrides
+ * can read broker-side feature flags (e.g. whether a
streams topology plugin is
+ * configured).
*/
- default boolean shouldExpire() {
+ default boolean shouldExpire(GroupCoordinatorConfig config) {
return true;
}
}
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 579e9726f93..e7cc6bd5339 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
@@ -862,7 +862,7 @@ public class GroupCoordinatorService implements
GroupCoordinator {
// 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());
+ Map<String, Integer> toClear = new
HashMap<>(eligible.size());
eligible.forEach((groupId, expectedStoredEpoch) -> {
if (failures.containsKey(groupId)) {
// Plugin failed: leave both stored epoch and
the push-path
@@ -879,9 +879,13 @@ public class GroupCoordinatorService implements
GroupCoordinator {
// 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));
+ toClear.put(groupId, expectedStoredEpoch);
});
- return
CompletableFuture.allOf(clearFutures.toArray(new CompletableFuture<?>[0]));
+ if (toClear.isEmpty()) return
CompletableFuture.completedFuture(null);
+ // All groups in `eligible` came from the same
partition's read so they
+ // hash to the same __consumer_offsets partition; one
batched write covers
+ // every clear on this shard.
+ return
clearStoredDescriptionTopologyEpochBatchAsync(toClear);
}));
return null;
}));
@@ -924,22 +928,28 @@ public class GroupCoordinatorService implements
GroupCoordinator {
}
/**
- * 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.
+ * Batched conditional metadata write that clears {@code
StoredDescriptionTopologyEpoch}
+ * for every entry in {@code expectedStoredEpochByGroupId}, but only for
the entries whose
+ * persisted value still equals the supplied epoch. Mismatches and missing
groups are
+ * silently ignored by the shard-side method. All groups in the batch must
hash to the same
+ * __consumer_offsets partition (the caller guarantees this — the
eligibility scan is per
+ * partition). 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(
+ private CompletableFuture<Void>
clearStoredDescriptionTopologyEpochBatchAsync(
+ Map<String, Integer> expectedStoredEpochByGroupId
+ ) {
+ if (expectedStoredEpochByGroupId.isEmpty()) return
CompletableFuture.completedFuture(null);
+ TopicPartition tp =
topicPartitionFor(expectedStoredEpochByGroupId.keySet().iterator().next());
+ return runtime.scheduleWriteOperation(
"clear-stored-topology-epoch",
- topicPartitionFor(groupId),
- coordinator ->
coordinator.clearStoredDescriptionTopologyEpoch(groupId, expectedStoredEpoch)
+ tp,
+ coordinator ->
coordinator.clearStoredDescriptionTopologyEpochBatch(expectedStoredEpochByGroupId)
).handle((__, throwable) -> {
if (throwable != null) {
- log.warn("Failed to clear StoredDescriptionTopologyEpoch for
group {}; the next cleanup cycle will retry.",
- groupId, throwable);
+ log.warn("Failed to clear StoredDescriptionTopologyEpoch for
groups {} on partition {}; the next cleanup cycle will retry.",
+ expectedStoredEpochByGroupId.keySet(), tp, throwable);
}
return null;
});
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 b79f7ebe3a4..8a0c2b508ba 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
@@ -985,26 +985,27 @@ public class GroupCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
/**
* 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
+ * (no live members), no committed offsets and no pending transactional
offsets, 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>This sweep relies on the regular offset-expiration cycle to have
already tombstoned
+ * expired offsets, so the eligibility check itself only asks "does this
group still have any
+ * offsets at all?" — see {@link OffsetMetadataManager#groupHasNoOffsets}.
Three independent
+ * cycles thus compose cleanly: offset expiration, topology cleanup
(here), and group
+ * tombstoning on the next sweep once the stored epoch is cleared.
+ *
* <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.
+ * topology epoch, and the no-offsets check.
*
* <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 {
@@ -1014,7 +1015,7 @@ public class GroupCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
if (!streamsGroup.isEmpty(committedOffset)) continue;
int storedEpoch =
streamsGroup.storedDescriptionTopologyEpoch(committedOffset);
if (storedEpoch == -1) continue;
- if (!offsetMetadataManager.allOffsetsExpired(groupId, now,
committedOffset)) continue;
+ if (!offsetMetadataManager.groupHasNoOffsets(groupId,
committedOffset)) continue;
eligible.put(groupId, storedEpoch);
} catch (Throwable t) {
// One bad group must not abort the whole scan; the next cycle
retries.
@@ -1037,6 +1038,28 @@ public class GroupCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
return
groupMetadataManager.clearStoredDescriptionTopologyEpoch(groupId,
expectedStoredEpoch);
}
+ /**
+ * Batched form of {@link #clearStoredDescriptionTopologyEpoch}: emits one
conditional clear
+ * record per entry in {@code expectedStoredEpochByGroupId} in a single
+ * {@link CoordinatorResult}, so the cycle issues one {@code
scheduleWriteOperation} per
+ * shard instead of one per group. Groups whose stored epoch no longer
matches yield no
+ * record (the GMM-level method returns an empty list for those), matching
the single-group
+ * variant's silent skip.
+ *
+ * <p>The result is non-atomic ({@code isAtomic=false}): the conditional
clear records are
+ * independent per group, so the runtime is free to split them across
multiple log batches
+ * if a single batch would exceed the limit. Any per-record write failure
is retried
+ * naturally on the next cycle because the persisted storedEpoch is still
non-default.
+ */
+ public CoordinatorResult<Void, CoordinatorRecord>
clearStoredDescriptionTopologyEpochBatch(
+ Map<String, Integer> expectedStoredEpochByGroupId
+ ) {
+ List<CoordinatorRecord> records = new
ArrayList<>(expectedStoredEpochByGroupId.size());
+ expectedStoredEpochByGroupId.forEach((groupId, expectedStoredEpoch) ->
+
records.addAll(groupMetadataManager.clearStoredDescriptionTopologyEpoch(groupId,
expectedStoredEpoch).records()));
+ return new CoordinatorResult<>(records, false);
+ }
+
/**
* Handles a ShareGroupDescribe request.
*
@@ -1109,14 +1132,17 @@ 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()) return;
+ // Offset cleanup runs for every group regardless of whether
shouldExpire(config)
+ // returns true — see Group.shouldExpire's javadoc — so a group
whose metadata is
+ // pinned (e.g. a streams group with a pending topology-plugin
clean-up) still has
+ // its expired committed offsets tombstoned, which is what feeds
the topology
+ // cleanup cycle's "no offsets remaining" eligibility check.
boolean allOffsetsExpired =
offsetMetadataManager.cleanupExpiredOffsets(groupId, records);
if (!allOffsetsExpired) return;
- if
(deferStreamsGroupTombstoneForPluginCleanup(topologyPluginConfigured, group))
return;
+ if (!group.shouldExpire(config)) return;
groupMetadataManager.maybeDeleteGroup(groupId, records);
});
@@ -1130,27 +1156,6 @@ 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/OffsetMetadataManager.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java
index 1465af9e8e4..7ef7b989fc6 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,7 +57,6 @@ 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;
@@ -282,21 +281,6 @@ 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.
*
@@ -1061,55 +1045,21 @@ public class OffsetMetadataManager {
}
/**
- * 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}.
+ * Whether {@code groupId} currently has no committed offsets and no
pending transactional
+ * offsets at the snapshot {@code committedOffset}. Used by the
topology-description plugin
+ * cleanup cycle on the eligibility read side — the regular
offset-expiration cycle is
+ * already running periodically and tombstoning expired offsets, so by the
time this is
+ * called the group's {@code offsetsByGroup} entry is gone iff every
committed offset has
+ * been expired and no transactional commits are in flight.
*
* <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.
+ * that calls this method. Both lookups go through 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);
+ public boolean groupHasNoOffsets(String groupId, long committedOffset) {
+ return offsets.offsetsByGroup.get(groupId, committedOffset) == null
+ && !openTransactions.contains(groupId, committedOffset);
}
/**
@@ -1143,6 +1093,7 @@ 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)) {
appendOffsetCommitTombstone(groupId, topic, partition,
records);
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java
index 290bbbdc416..1029f32ae3d 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java
@@ -25,6 +25,7 @@ import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.coordinator.common.runtime.CoordinatorMetadataImage;
import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
import org.apache.kafka.coordinator.group.CommitPartitionValidator;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
import org.apache.kafka.coordinator.group.GroupCoordinatorRecordHelpers;
import org.apache.kafka.coordinator.group.OffsetExpirationCondition;
import org.apache.kafka.coordinator.group.modern.ModernGroup;
@@ -325,7 +326,9 @@ public class ShareGroup extends
ModernGroup<ShareGroupMember> {
}
@Override
- public boolean shouldExpire() {
+ public boolean shouldExpire(GroupCoordinatorConfig config) {
+ // Share groups don't expose committed-offset semantics, so the
natural-expiration sweep
+ // is never used to delete them.
return false;
}
}
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 ac5219a0793..5f392e5a698 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
@@ -31,6 +31,7 @@ import
org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
import org.apache.kafka.coordinator.common.runtime.CoordinatorTimer;
import org.apache.kafka.coordinator.group.CommitPartitionValidator;
import org.apache.kafka.coordinator.group.Group;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
import org.apache.kafka.coordinator.group.OffsetExpirationCondition;
import org.apache.kafka.coordinator.group.OffsetExpirationConditionImpl;
import org.apache.kafka.coordinator.group.TargetAssignmentMetadata;
@@ -757,6 +758,24 @@ public class StreamsGroup implements Group {
return storedDescriptionTopologyEpoch.get();
}
+ /**
+ * Defer tombstoning of an empty streams group while the broker-level
topology-description
+ * cleanup cycle still has work to do for it: a plugin is configured and
the persisted
+ * {@code StoredDescriptionTopologyEpoch} is not the {@code -1} default.
The cycle drives
+ * {@code plugin.deleteTopology} and clears the stored epoch; the next
sweep then proceeds
+ * with tombstoning. When no plugin is configured the gate never holds, so
deferring
+ * indefinitely is not possible.
+ *
+ * <p>Per the {@link Group#shouldExpire(GroupCoordinatorConfig)} contract
this only gates the
+ * group-metadata tombstone — committed offsets are still expired by the
sweep regardless of
+ * what this returns, which is what feeds the topology-cleanup cycle's
eligibility check.
+ */
+ @Override
+ public boolean shouldExpire(GroupCoordinatorConfig config) {
+ return !(config.isStreamsGroupTopologyDescriptionPluginConfigured()
+ && storedDescriptionTopologyEpoch() != -1);
+ }
+
/**
* @param committedOffset A committed offset corresponding to the desired
snapshot.
* @return The topology epoch most recently successfully stored by the
topology description plugin at the given
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 668b7011590..4edd2598550 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
@@ -51,6 +51,7 @@ import org.mockito.MockedStatic;
import java.time.Duration;
import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -965,6 +966,32 @@ public class
GroupCoordinatorServiceTopologyDescriptionTest {
verify(runtime,
times(1)).scheduleWriteOperation(eq("clear-stored-topology-epoch"),
eq(GROUP_TP), any());
}
+ @Test
+ public void testCleanupCycleBatchesClearWritesPerPartition() {
+ // Two eligible groups land on the same partition's eligibility read —
they must trigger
+ // exactly one scheduleWriteOperation carrying both conditional
clears, not one write per
+ // group.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+
when(plugin.deleteTopology("foo")).thenReturn(CompletableFuture.completedFuture(null));
+
when(plugin.deleteTopology("bar")).thenReturn(CompletableFuture.completedFuture(null));
+ Map<String, Integer> eligible = new LinkedHashMap<>();
+ eligible.put("foo", 4);
+ eligible.put("bar", 9);
+
when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"),
any()))
+ .thenReturn(List.of(CompletableFuture.completedFuture(eligible)));
+ when(runtime.scheduleWriteOperation(eq("clear-stored-topology-epoch"),
any(), any()))
+ .thenReturn(CompletableFuture.completedFuture(null));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ service.runOneStreamsTopologyCleanupCycle();
+
+ verify(plugin, times(1)).deleteTopology("foo");
+ verify(plugin, times(1)).deleteTopology("bar");
+ // One write covers both groups; not two per-group writes.
+ verify(runtime,
times(1)).scheduleWriteOperation(eq("clear-stored-topology-epoch"), any(),
any());
+ }
+
@Test
public void testCleanupCycleSkipsClearOnPluginFailure() {
// Plugin fails -> the cycle must NOT clear stored epoch; the group
stays gated on
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 ca005188ef9..80340c376f2 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
@@ -115,6 +115,7 @@ import org.mockito.Mockito;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -1525,8 +1526,9 @@ public class GroupCoordinatorShardTest {
// - 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
+ // - !groupHasNoOffsets -> skip (offsets still present; the regular
expiration
+ // cycle will tombstone them and the next sweep here will pick the
group up)
+ // - empty + stored != -1 + no offsets -> include, value = observed
storedEpoch
GroupMetadataManager groupMetadataManager =
mock(GroupMetadataManager.class);
OffsetMetadataManager offsetMetadataManager =
mock(OffsetMetadataManager.class);
GroupCoordinatorConfig config = mock(GroupCoordinatorConfig.class);
@@ -1576,7 +1578,7 @@ public class GroupCoordinatorShardTest {
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);
+ when(offsetMetadataManager.groupHasNoOffsets(eq("unexpired-offsets"),
anyLong())).thenReturn(false);
// eligible: passes every predicate; storedEpoch=7 must appear in the
result map.
StreamsGroup eligible = mock(StreamsGroup.class);
@@ -1584,7 +1586,7 @@ public class GroupCoordinatorShardTest {
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);
+ when(offsetMetadataManager.groupHasNoOffsets(eq("eligible"),
anyLong())).thenReturn(true);
Map<String, Integer> result =
coordinator.listStreamsGroupsNeedingTopologyCleanup(committedOffset);
@@ -1626,7 +1628,7 @@ public class GroupCoordinatorShardTest {
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);
+ when(offsetMetadataManager.groupHasNoOffsets(eq("good"),
anyLong())).thenReturn(true);
Map<String, Integer> result =
coordinator.listStreamsGroupsNeedingTopologyCleanup(committedOffset);
@@ -1664,17 +1666,16 @@ public class GroupCoordinatorShardTest {
}
@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.
+ public void
testClearStoredDescriptionTopologyEpochBatchConcatenatesGMMRecords() {
+ // The batch shard method must invoke the GMM's single-group clear
once per entry and
+ // fold the resulting records into one CoordinatorResult. The cycle
relies on this so a
+ // partition with N eligible groups produces one
scheduleWriteOperation carrying N
+ // conditional clear records, not N separate writes.
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();
+ MockTime mockTime = new MockTime();
MockCoordinatorTimer<CoordinatorRecord> timer = new
MockCoordinatorTimer<>(mockTime);
GroupCoordinatorShard coordinator = new GroupCoordinatorShard(
new LogContext(),
@@ -1687,41 +1688,41 @@ public class GroupCoordinatorShardTest {
mock(CoordinatorMetricsShard.class)
);
- CoordinatorRecord offsetCommitTombstone =
GroupCoordinatorRecordHelpers.newOffsetCommitTombstoneRecord("group-id",
"topic", 0);
+ CoordinatorRecord recordA = mock(CoordinatorRecord.class);
+ CoordinatorRecord recordB = mock(CoordinatorRecord.class);
+ // "matched" group: GMM returns one record. "mismatched" group: GMM
returns empty, so
+ // the batched result still contains only the matched group's record.
+
when(groupMetadataManager.clearStoredDescriptionTopologyEpoch("matched", 3))
+ .thenReturn(new CoordinatorResult<>(List.of(recordA)));
+
when(groupMetadataManager.clearStoredDescriptionTopologyEpoch("mismatched", 5))
+ .thenReturn(new CoordinatorResult<>(List.of()));
+
when(groupMetadataManager.clearStoredDescriptionTopologyEpoch("matched-2", 4))
+ .thenReturn(new CoordinatorResult<>(List.of(recordB)));
- @SuppressWarnings("unchecked")
- ArgumentCaptor<List<CoordinatorRecord>> recordsCapture =
ArgumentCaptor.forClass(List.class);
+ Map<String, Integer> input = new LinkedHashMap<>();
+ input.put("matched", 3);
+ input.put("mismatched", 5);
+ input.put("matched-2", 4);
- StreamsGroup streamsGroup = mock(StreamsGroup.class);
- when(streamsGroup.shouldExpire()).thenReturn(true);
- when(streamsGroup.type()).thenReturn(Group.GroupType.STREAMS);
- when(streamsGroup.storedDescriptionTopologyEpoch()).thenReturn(4);
+ CoordinatorResult<Void, CoordinatorRecord> result =
coordinator.clearStoredDescriptionTopologyEpochBatch(input);
- 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());
+ assertEquals(List.of(recordA, recordB), result.records());
+ // Non-atomic: the per-group conditional clears are independent, so
the runtime is free
+ // to split them across multiple log batches if a single batch would
exceed the limit.
+ assertFalse(result.isAtomic());
+
verify(groupMetadataManager).clearStoredDescriptionTopologyEpoch("matched", 3);
+
verify(groupMetadataManager).clearStoredDescriptionTopologyEpoch("mismatched",
5);
+
verify(groupMetadataManager).clearStoredDescriptionTopologyEpoch("matched-2",
4);
}
@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.
+ public void
testCleanupGroupMetadataKeepsOffsetTombstonesWhenGroupRefusesToExpire() {
+ // When a group's shouldExpire(config) returns false after offset
cleanup has run, the
+ // shard sweep must skip maybeDeleteGroup and still keep the offset
tombstones it
+ // accumulated. The decision lives on the group; the shard only honors
it.
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);
@@ -1736,29 +1737,37 @@ public class GroupCoordinatorShardTest {
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);
+ CoordinatorRecord offsetCommitTombstone =
GroupCoordinatorRecordHelpers.newOffsetCommitTombstoneRecord("group-id",
"topic", 0);
+
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<List<CoordinatorRecord>> recordsCapture =
ArgumentCaptor.forClass(List.class);
+
+ Group group = mock(Group.class);
+ when(group.shouldExpire(config)).thenReturn(false);
when(groupMetadataManager.groupIds()).thenReturn(Set.of("group-id"));
- when(groupMetadataManager.group("group-id")).thenReturn(streamsGroup);
- when(offsetMetadataManager.cleanupExpiredOffsets(eq("group-id"),
any())).thenReturn(true);
+ when(groupMetadataManager.group("group-id")).thenReturn(group);
+ when(offsetMetadataManager.cleanupExpiredOffsets(eq("group-id"),
recordsCapture.capture()))
+ .thenAnswer(invocation -> {
+ recordsCapture.getValue().add(offsetCommitTombstone);
+ return true;
+ });
- coordinator.cleanupGroupMetadata();
+ CoordinatorResult<Void, CoordinatorRecord> result =
coordinator.cleanupGroupMetadata();
- verify(groupMetadataManager,
times(1)).maybeDeleteGroup(eq("group-id"), any());
+ // 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
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.
+ public void testCleanupGroupMetadataTombstonesWhenGroupAgreesToExpire() {
+ // When the group's shouldExpire(config) returns true after offset
cleanup has run,
+ // the shard sweep proceeds to maybeDeleteGroup.
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);
@@ -1773,12 +1782,11 @@ public class GroupCoordinatorShardTest {
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.
+ Group group = mock(Group.class);
+ when(group.shouldExpire(config)).thenReturn(true);
+
when(groupMetadataManager.groupIds()).thenReturn(Set.of("group-id"));
- when(groupMetadataManager.group("group-id")).thenReturn(streamsGroup);
+ when(groupMetadataManager.group("group-id")).thenReturn(group);
when(offsetMetadataManager.cleanupExpiredOffsets(eq("group-id"),
any())).thenReturn(true);
coordinator.cleanupGroupMetadata();
@@ -1808,6 +1816,11 @@ public class GroupCoordinatorShardTest {
when(groupMetadataManager.groupIds()).thenReturn(Set.of("group-id"));
when(groupMetadataManager.group("group-id")).thenReturn(group);
+ // Offset cleanup runs uniformly; for share groups it has no committed
offsets and no
+ // pending transactions, so cleanupExpiredOffsets returns true. The
sweep then consults
+ // shouldExpire(config), which ShareGroup overrides to return false —
the group-metadata
+ // tombstone is suppressed.
+ when(offsetMetadataManager.cleanupExpiredOffsets(eq("group-id"),
any())).thenReturn(true);
assertFalse(timer.contains(GROUP_EXPIRATION_KEY));
CoordinatorResult<Void, CoordinatorRecord> result =
coordinator.cleanupGroupMetadata();
@@ -1819,7 +1832,7 @@ public class GroupCoordinatorShardTest {
assertNull(result.appendFuture());
verify(groupMetadataManager, times(1)).groupIds();
- verify(offsetMetadataManager,
times(0)).cleanupExpiredOffsets(eq("group-id"), any());
+ verify(offsetMetadataManager,
times(1)).cleanupExpiredOffsets(eq("group-id"), any());
verify(groupMetadataManager,
times(0)).maybeDeleteGroup(eq("group-id"), any());
}
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 960224f7156..4c32a718a5c 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,8 +104,6 @@ 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;
@@ -340,11 +338,11 @@ public class OffsetMetadataManagerTest {
return isOffsetsEmptyForGroup;
}
- public boolean allOffsetsExpired(String groupId, long
currentTimestampMs) {
+ public boolean groupHasNoOffsets(String groupId) {
// 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);
+ return offsetMetadataManager.groupHasNoOffsets(groupId,
Long.MAX_VALUE);
}
public List<OffsetFetchResponseData.OffsetFetchResponseTopics>
fetchOffsets(
@@ -3259,146 +3257,33 @@ public class OffsetMetadataManagerTest {
}
@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.
+ public void testGroupHasNoOffsetsReturnsTrueWhenGroupAbsentAndNoOpenTxn() {
+ // Unknown group with no entry in offsetsByGroup and no open
transactions: the
+ // topology-cleanup cycle is free to call plugin.deleteTopology and
clear the stored
+ // epoch. Mirrors the post-state the offset-expiration cycle leaves
behind once it has
+ // tombstoned the last committed offset for a group.
OffsetMetadataManagerTestContext context = new
OffsetMetadataManagerTestContext.Builder().build();
- assertTrue(context.allOffsetsExpired("unknown-group-id",
context.time.milliseconds()));
+ assertTrue(context.groupHasNoOffsets("unknown-group-id"));
}
@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();
-
+ public void testGroupHasNoOffsetsReturnsFalseWhenOpenTxnPending() {
// A pending transactional commit creates an openTransactions entry
without populating
- // the durable offsets map.
+ // the durable offsets map. Even though offsetsByGroup is null, the
open txn could land
+ // a fresh offset on commit — the topology cycle must hold off.
+ OffsetMetadataManagerTestContext context = new
OffsetMetadataManagerTestContext.Builder().build();
context.commitOffset(42L, "group-id", "foo", 0, 100L, 0,
context.time.milliseconds());
- assertFalse(context.allOffsetsExpired("group-id",
context.time.milliseconds()));
+ assertFalse(context.groupHasNoOffsets("group-id"));
}
@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();
-
+ public void testGroupHasNoOffsetsReturnsFalseWhenCommittedOffsetExists() {
+ // Group still has a durable committed offset: the offset-expiration
cycle has not
+ // tombstoned it yet (either still inside the retention window or held
by an active
+ // subscription), so the topology cycle is not eligible to fire.
+ OffsetMetadataManagerTestContext context = new
OffsetMetadataManagerTestContext.Builder().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()));
+ assertFalse(context.groupHasNoOffsets("group-id"));
}
private static OffsetFetchResponseData.OffsetFetchResponsePartitions
mkOffsetPartitionResponse(
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java
index 7b0e97a8bcb..a75c1083149 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java
@@ -34,6 +34,7 @@ import
org.apache.kafka.coordinator.common.runtime.CoordinatorTimer;
import
org.apache.kafka.coordinator.common.runtime.KRaftCoordinatorMetadataImage;
import org.apache.kafka.coordinator.common.runtime.MetadataImageBuilder;
import org.apache.kafka.coordinator.group.CommitPartitionValidator;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
import org.apache.kafka.coordinator.group.OffsetAndMetadata;
import org.apache.kafka.coordinator.group.OffsetExpirationCondition;
import org.apache.kafka.coordinator.group.OffsetExpirationConditionImpl;
@@ -1416,4 +1417,43 @@ public class StreamsGroupTest {
assertTrue(streamsGroup.cachedEndpointToPartitions("member-1").isEmpty());
assertTrue(streamsGroup.cachedEndpointToPartitions("member-2").isEmpty());
}
+
+ @Test
+ public void testShouldExpireFalseWhenPluginConfiguredAndEpochStored() {
+ // Defer holds: a topology plugin is configured on this broker and the
group has a
+ // non-default stored epoch — the natural-expiration sweep must wait
for the broker-level
+ // topology-cleanup cycle to drive plugin.deleteTopology and clear the
field before the
+ // group is tombstoned.
+ StreamsGroup streamsGroup = createStreamsGroup("group-id");
+ streamsGroup.setStoredDescriptionTopologyEpoch(4);
+ GroupCoordinatorConfig config = mock(GroupCoordinatorConfig.class);
+
when(config.isStreamsGroupTopologyDescriptionPluginConfigured()).thenReturn(true);
+
+ assertFalse(streamsGroup.shouldExpire(config));
+ }
+
+ @Test
+ public void testShouldExpireTrueWhenStoredEpochIsDefault() {
+ // storedDescriptionTopologyEpoch == -1: no plugin row exists for this
group (either the
+ // topology cleanup cycle has already cleared it on a previous tick,
or one was never
+ // pushed). The tombstone proceeds.
+ StreamsGroup streamsGroup = createStreamsGroup("group-id");
+ GroupCoordinatorConfig config = mock(GroupCoordinatorConfig.class);
+
when(config.isStreamsGroupTopologyDescriptionPluginConfigured()).thenReturn(true);
+
+ assertTrue(streamsGroup.shouldExpire(config));
+ }
+
+ @Test
+ public void testShouldExpireTrueWhenNoPluginConfigured() {
+ // No plugin configured on this broker: no topology cleanup cycle will
ever run to clear
+ // the field, so blocking the tombstone would prevent natural
expiration indefinitely.
+ // The stored epoch value is irrelevant in this branch.
+ StreamsGroup streamsGroup = createStreamsGroup("group-id");
+ streamsGroup.setStoredDescriptionTopologyEpoch(4);
+ GroupCoordinatorConfig config = mock(GroupCoordinatorConfig.class);
+
when(config.isStreamsGroupTopologyDescriptionPluginConfigured()).thenReturn(false);
+
+ assertTrue(streamsGroup.shouldExpire(config));
+ }
}
