lucasbru commented on code in PR #22622:
URL: https://github.com/apache/kafka/pull/22622#discussion_r3453592271
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java:
##########
@@ -1034,6 +1060,78 @@ public OffsetFetchResponseData.OffsetFetchResponseGroup
fetchAllOffsets(
.setTopics(topicResponses);
}
+ /**
+ * Whether the offset committed for {@code (groupId, topic, partition)} is
currently
+ * eligible for expiration: its age exceeds {@code offsets.retention.ms}
per the group's
+ * {@link OffsetExpirationCondition}, and there is no pending
transactional offset on the
+ * partition. Shared between {@link #cleanupExpiredOffsets} (the write
path that tombstones
+ * eligible offsets) and {@link #allOffsetsExpired} (the read-only check
used by the
+ * topology-description cleanup cycle) so the two cannot drift.
Review Comment:
`allOffsetsExpired` does not call this helper — it inlines its own
per-partition check using the snapshot-aware
`openTransactions.contains(groupId, topic, partition, committedOffset)`, while
this method uses non-snapshot `hasPendingTransactionalOffsets`. The two paths
must diverge on snapshot semantics, so there is no shared abstraction here.
Worth reverting the extraction: remove `isOffsetExpirable`, put the condition
back inline in `cleanupExpiredOffsets`, and let each method own its check
explicitly.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java:
##########
@@ -3249,6 +3258,119 @@ public void
testCleanupExpiredOffsetsWithPendingTransactionalOffsetsOnly() {
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
Review Comment:
This tests the per-partition pending-txn branch (same partition has both a
committed offset and a pending transactional offset). Missing a case for the
trailing `return !openTransactions.contains(groupId, committedOffset)` when
`offsetsByTopic` is non-null: committed offsets for partition A are all past
retention, but there is an open transaction on partition B with no committed
offset at all. `allOffsetsExpired` should return false there; the trailing
group-level check is the only thing that catches it.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -786,6 +796,26 @@ private void
throwIfStreamsGroupTopologyDescriptionUpdateInvalid(
throwIfNull(request.topologyDescription(), "TopologyDescription can't
be null.");
}
+ /**
+ * Record per-call outcomes from a batched {@code plugin.deleteTopology}
invocation
+ * triggered by an explicit {@code DeleteGroups}. The periodic cleanup
cycle has its
+ * own recorder inside {@link StreamsGroupTopologyDescriptionManager};
both write to
+ * the same {@code delete-success} / {@code delete-error} sensors so a
single pair of
+ * meters tracks every {@code plugin.deleteTopology} the broker drives,
regardless of
+ * trigger.
+ */
+ private void recordPluginDeleteOutcome(int attempted, int errors) {
Review Comment:
This is a duplicate and should probably go into the manager
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -2627,6 +2658,10 @@ public void startup(
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 captures the runtime via the
timer task's lambda
+ // rather than holding it as a field, so its lifecycle stays decoupled
from the runtime.
Review Comment:
"rather than holding it as a field" explains why this design was chosen
relative to an alternative — that belongs in the PR description, not an inline
comment. The comment could just say what the invariant is: the manager doesn't
retain a runtime reference past `close()`.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java:
##########
@@ -1514,6 +1517,244 @@ public void testCleanupGroupMetadataForConsumerGroup() {
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
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
Review Comment:
No test for `GroupCoordinatorShard.clearStoredDescriptionTopologyEpoch`. The
GMM-level tests cover epoch-match, epoch-mismatch, and missing-group, but the
shard delegation itself is not exercised. Given
`listStreamsGroupsNeedingTopologyCleanup` gets its own shard test, this should
too.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java:
##########
@@ -68,29 +80,278 @@ public class StreamsGroupTopologyDescriptionManager
implements AutoCloseable {
private final Optional<StreamsGroupTopologyDescriptionPlugin> plugin;
private final StreamsGroupTopologyDescriptionBackoff backoff;
+ private final Timer timer;
+ private final long cleanupCheckIntervalMs;
+ private final Function<String, TopicPartition> topicPartitionFor;
+ private final GroupCoordinatorMetrics groupCoordinatorMetrics;
+
+ /**
+ * True between {@link #startCleanupCycle} and {@link #close}. Read at
every cycle
+ * boundary that would otherwise schedule new plugin calls or runtime
writes, so a
+ * cycle that is in flight when {@code close} fires drains rather than
racing the
+ * runtime tear-down.
+ */
+ 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 per-group plugin calls and
conditional-clear writes
+ * is dropped. Set true at the top of {@link #runCleanupCycle}, released
by the
+ * terminal {@code whenComplete} that joins all per-partition reads and
per-group
+ * futures.
+ */
+ 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,
- Time time
+ Time time,
+ Timer timer,
+ long cleanupCheckIntervalMs,
Review Comment:
The manager now takes `timer`, `cleanupCheckIntervalMs`,
`topicPartitionFor`, and `groupCoordinatorMetrics` — all service-level concerns
— plus receives `runtime` via `startCleanupCycle`. This pushes coordinator
infrastructure knowledge into a class whose original job was plugin lifecycle
and per-group backoff.
A cleaner split: keep cycle *orchestration* in the service and give
`startCleanupCycle` a `Supplier<CompletableFuture<?>>` instead of `runtime`:
```java
// service side
manager.startCleanupCycle(timer, intervalMs, () ->
runOneCleanupCycle(runtime));
private CompletableFuture<?> runOneCleanupCycle(runtime) {
// scheduleReadAllOperation, invokeDeleteTopologies,
scheduleWriteOperation
// service already owns topicPartitionFor and groupCoordinatorMetrics
}
```
The manager then only needs to: fire the supplier on the timer, guard with
`cycleInFlight`, and flip `running`. Constructor goes back to `(logContext,
plugin, time)`. `topicPartitionFor`, `GroupCoordinatorMetrics`, and the
`runtime` import all stay on the service side where they already live.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
