jeffkbkim commented on code in PR #14467:
URL: https://github.com/apache/kafka/pull/14467#discussion_r1350863763
##########
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:
> In this case, isn't subscribedTopics going to be set to an optional
containing an empty list?
ah.. i misread and thought subscribedTopics becomes an empty optional, not
an empty set. So we would return `false` which is the correct behavior.
> Did you do it because it is correct or by mistake?
which change, to add another argument? I added that so we can rely on using
`isSubscribedToTopic` instead of passing in the subscribed topics set.
i will update with your suggestion, it makes sense to me.
##########
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:
> In this case, isn't subscribedTopics going to be set to an optional
containing an empty list?
ah.. i misread and thought subscribedTopics becomes an empty optional, not
an empty set. So we would return `false` which is the correct behavior.
> Did you do it because it is correct or by mistake?
which change, to add another argument? I added that so we can rely on using
`isSubscribedToTopic` instead of passing in the subscribed topics set.
i will update with your suggestion, it makes sense to me.
--
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
