dajac commented on code in PR #14467:
URL: https://github.com/apache/kafka/pull/14467#discussion_r1350261109
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java:
##########
@@ -544,6 +573,81 @@ public OffsetFetchResponseData.OffsetFetchResponseGroup
fetchAllOffsets(
.setTopics(topicResponses);
}
+ /**
+ * Remove expired offsets for group.
+ *
+ * @param groupId The group id.
+ * @param records The list of records to populate with offset commit
tombstone records.
+ * @param offsetsRetentionMs The offset retention in milliseconds.
+ *
+ * @return The group id if the group no longer has any offsets remaining,
empty otherwise.
+ */
+ public Optional<String> cleanupExpiredOffsets(String groupId, List<Record>
records, long offsetsRetentionMs) {
+ TimelineHashMap<String, TimelineHashMap<Integer, OffsetAndMetadata>>
offsetsByTopic = offsetsByGroup.get(groupId);
+ if (offsetsByTopic == null) {
+ return Optional.of(groupId);
+ }
+ try {
+ Group group = groupMetadataManager.group(groupId);
+ ExpirationCondition expirationCondition =
group.expirationCondition();
+ Set<TopicPartition> expiredPartitions = new HashSet<>();
+ long currentTimestamp = time.milliseconds();
+ AtomicBoolean hasAllOffsetsExpired = new AtomicBoolean(true);
+ offsetsByTopic.forEach((topic, partitions) -> {
+ if (!expirationCondition.subscribedTopics.contains(topic)) {
Review Comment:
Hum.. I am not sure to follow. In this case, isn't `subscribedTopics` going
to be set to an optional containing an empty list?
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
