AndrewJSchofield commented on code in PR #21045:
URL: https://github.com/apache/kafka/pull/21045#discussion_r2582682164
##########
server/src/main/java/org/apache/kafka/server/share/session/ShareSessionCache.java:
##########
@@ -146,32 +166,48 @@ public synchronized void
maybeRemoveAndNotifyListeners(ShareSessionKey key) {
// update the evictions metric.
evictionsMeter.mark();
// Try removing session if not already removed. The listener might
have removed the session
- // already.
- remove(session);
+ // already. Don't notify the share group listener as it should be
notified later. Notify
+ // the listener only once per removal.
+ removeAndMaybeNotifyListener(session, false);
}
// Notify the share group listener if the group is empty. This should
be checked regardless
// session is evicted by connection disconnect or client's final epoch.
- int numMembers = numMembersPerGroup.getOrDefault(key.groupId(), 0);
+ checkAndNotifyGroupListener(key.groupId());
+ }
+
+ /**
+ * Check if the share group is empty and notify the share group listener.
+ *
+ * @param groupId The share group id.
+ */
+ private synchronized void checkAndNotifyGroupListener(String groupId) {
+ int numMembers = numMembersPerGroup.getOrDefault(groupId, 0);
if (numMembers == 0) {
// Remove the group from the map as it is empty.
- numMembersPerGroup.remove(key.groupId());
+ numMembersPerGroup.remove(groupId);
if (shareGroupListener != null) {
- shareGroupListener.onGroupEmpty(key.groupId());
+ shareGroupListener.onGroupEmpty(groupId);
Review Comment:
Are you sure that you want to call into the replica manager while within the
synchronized block for the share-session cache? I wonder whether we are heading
towards a surprising deadlock.
--
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]
