adixitconfluent commented on code in PR #21062:
URL: https://github.com/apache/kafka/pull/21062#discussion_r2585099717
##########
server/src/main/java/org/apache/kafka/server/share/session/ShareSessionCache.java:
##########
@@ -172,6 +169,17 @@ public synchronized ShareSession remove(ShareSession
session) {
if (removeResult != null) {
numPartitions = numPartitions - session.cachedSize();
numMembersPerGroup.compute(session.key().groupId(), (k, v) -> v !=
null ? v - 1 : 0);
+ // Mark the session as stale in the connectionIdToSessionMap to
avoid removing
+ // the active session for the client. When client re-sends the
initial epoch where the
+ // broker removes the prior session and establishes new session,
then sometimes the connection
+ // id is changed. This leads to removal of the new session from
the cache, when old connection
+ // disconnect event is processed. Marking the old connection as
stale avoids this issue.
+ // If the connection id remains same for both old and new session,
then the subsequent call
+ // for session creation will overwrite the prior stale mapping in
the connectionIdToSessionMap.
Review Comment:
ahh, I think you mean after disconnection
--
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]
