chirag-wadhwa5 commented on code in PR #16792:
URL: https://github.com/apache/kafka/pull/16792#discussion_r1708756303
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -430,6 +430,41 @@ public ShareFetchContext newContext(String groupId,
Map<TopicIdPartition, ShareF
return context;
}
+ /**
+ * The acknowledgeSessionUpdate method is used to update the request epoch
and lastUsed time of the share session.
+ * @param groupId The group id in the share fetch request.
+ * @param reqMetadata The metadata in the share acknowledge request.
+ */
+ public void acknowledgeSessionUpdate(String groupId, ShareFetchMetadata
reqMetadata) {
+ if (reqMetadata.epoch() == ShareFetchMetadata.INITIAL_EPOCH) {
+ // ShareAcknowledge Request cannot have epoch as INITIAL_EPOCH (0)
+ throw Errors.INVALID_SHARE_SESSION_EPOCH.exception();
+ } else if (reqMetadata.epoch() == ShareFetchMetadata.FINAL_EPOCH) {
+ ShareSessionKey key = shareSessionKey(groupId,
reqMetadata.memberId());
+ if (cache.remove(key) != null) {
+ log.debug("Removed share session with key " + key);
+ }
Review Comment:
Thanks for the review. I think you are right. We should check for the
existence of the share sessions in the final epoch, not just in case of
shareAcknowledge, but also in case of shareFetch. In case of shareFetch we
might have piggybacked acknowledgements in the final epoch and if the session
has already been closed / evicted, the current code would allow the
acknowledgements to proceed. I will make the required changes both in
`SharePartitionManager.newContext` and
`SharePartitionManager.acknowledgeSessionUpdate`. I would also request
@AndrewJSchofield and @apoorvmittal10 to also validate my understanding, thanks
!
--
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
