Repository: apache/kafka
File: server/src/main/java/org/apache/kafka/server/share/session/ShareSessionCache.java (Java)

Code before:
    Integer numMembers(String groupId) {
        return numMembersPerGroup.get(groupId);
    }

    // Visible for testing.
    synchronized SessionKeyAndState connectionSessionKeyAndState(String connectionId) {
        return connectionIdToSessionMap.get(connectionId);
    }

    private final class ClientConnectionDisconnectListener implements ConnectionDisconnectListener {

        // When the client disconnects, the corresponding session should be removed from the cache.
        @Override
        public void onDisconnect(String connectionId) {
            SessionKeyAndState sessionKeyAndState = maybeRemoveConnectionFromSession(connectionId);
            // If the session is not stale, try removing the session and notify listeners.
            if (sessionKeyAndState != null) {
                if (!sessionKeyAndState.stale()) {
                    // Try removing session and notify listeners. The session might already be removed
                    // as part of final epoch from client, so we need to check if the session is still
                    // present in the cache.
                    maybeRemoveAndNotifyListeners(sessionKeyAndState.shareSessionKey());
                }
                // 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.
                checkAndNotifyGroupListeners(sessionKeyAndState.shareSessionKey().groupId());
            }
        }
    }

    /**
     * The class records the session key and tracks if the session is stale. The session is marked stale
     * when the session is removed from the cache prior to the client disconnect event.
     */
    // Visible for testing.
    static class SessionKeyAndState {
        private final ShareSessionKey shareSessionKey;
        private boolean stale;

        SessionKeyAndState(ShareSessionKey shareSessionKey) {
            this.shareSessionKey = shareSessionKey;
            this.stale = false;
        }

        ShareSessionKey shareSessionKey() {
            return shareSessionKey;
        }

        boolean stale() {
            return stale;
        }

Review comment (suggestion):
I suggest changing this method name to `checkAndNotifyListenersOnGroupEmpty`. Now the two different listener methods are called in different places, I think it would be helpful to differentiate them more.

Code after:
    Integer numMembers(String groupId) {
        return numMembersPerGroup.get(groupId);
    }

    // Visible for testing.
    synchronized SessionKeyAndState connectionSessionKeyAndState(String connectionId) {
        return connectionIdToSessionMap.get(connectionId);
    }

    private final class ClientConnectionDisconnectListener implements ConnectionDisconnectListener {

        // When the client disconnects, the corresponding session should be removed from the cache.
        @Override
        public void onDisconnect(String connectionId) {
            SessionKeyAndState sessionKeyAndState = maybeRemoveConnectionFromSession(connectionId);
            if (sessionKeyAndState != null) {
                // If the session is not stale, try removing the session and notify listeners.
                if (!sessionKeyAndState.stale()) {
                    // Try removing session and notify listeners. The session might already be removed
                    // as part of final epoch from client, so we need to check if the session is still
                    // present in the cache.
                    maybeRemoveAndNotifyListenersOnMemberLeave(sessionKeyAndState.shareSessionKey());
                }
                // 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.
                checkAndNotifyListenersOnGroupEmpty(sessionKeyAndState.shareSessionKey().groupId());
            }
        }
    }

    /**
     * The class records the session key and tracks if the session is stale. The session is marked stale
     * when the session is removed from the cache prior to the client disconnect event.
     */
    // Visible for testing.
    static class SessionKeyAndState {
        private final ShareSessionKey shareSessionKey;
        private boolean stale;

        SessionKeyAndState(ShareSessionKey shareSessionKey) {
            this.shareSessionKey = shareSessionKey;
            this.stale = false;
        }

        ShareSessionKey shareSessionKey() {
            return shareSessionKey;
        }

        boolean stale() {
            return stale;
        }