dajac commented on code in PR #22622:
URL: https://github.com/apache/kafka/pull/22622#discussion_r3466905977
##########
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:
This is quite expensive to run this. Have we considered just checking if the
group id has any offsets without applying any logic? The regular expiration
will take care of checking this.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java:
##########
@@ -1065,9 +1143,8 @@ public boolean cleanupExpiredOffsets(String groupId,
List<CoordinatorRecord> rec
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.
Review Comment:
Why have we removed this comment?
--
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]
