lucasbru commented on code in PR #22622:
URL: https://github.com/apache/kafka/pull/22622#discussion_r3467104557
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java:
##########
@@ -1034,6 +1060,58 @@ public OffsetFetchResponseData.OffsetFetchResponseGroup
fetchAllOffsets(
.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
Review Comment:
So we'd get another clean cycle
-> 1) sweep the offsets
-> 2) sweep the topology
-> 3) sweep the group
That would be a nice simplification!
--
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]
