@Override
    public Mono<Set<Long>> sendNotificationToLocalClients(
            TracingContext tracingContext,
            ByteBuf notificationData,
            Set<Long> recipientIds,
            Set<UserSessionId> excludedUserSessionIds,
            @Nullable DeviceType excludedDeviceType) {
        try {
            Validator.notNull(notificationData, "notificationData");
            Validator.notEmpty(recipientIds, "recipientIds");
            Validator.notNull(excludedUserSessionIds, "excludedUserSessionIds");
        } catch (ResponseException e) {
            return Mono.error(e);
        }
        // Prepare data
        int recipientCount = recipientIds.size();
        int halfRecipientCount = recipientCount >> 1;
        Set<Long> offlineRecipientIds =
                CollectionUtil.newConcurrentSetWithExpectedSize(halfRecipientCount);

        List<Mono<Void>> monos = new ArrayList<>(halfRecipientCount);

        // Send notification
        boolean hasExcludedUserSessionIds = !excludedUserSessionIds.isEmpty();
        for (Long recipientId : recipientIds) {
            UserSessionsManager userSessionsManager =
                    sessionService.getUserSessionsManager(recipientId);
            if (userSessionsManager == null) {
                offlineRecipientIds.add(recipientId);
            } else {
                for (UserSession userSession : userSessionsManager.getDeviceTypeToSession()
                        .values()) {
                    if (excludedDeviceType == userSession.getDeviceType()
                            || (hasExcludedUserSessionIds
                                    && excludedUserSessionIds.contains(new UserSessionId(
                                            userSession.getUserId(),
                                            userSession.getDeviceType())))) {
                        continue;
                    }
                    notificationData.retain();
                    // It is the responsibility of the downstream to decrease the reference count of
                    // the notification by 1
                    // when the notification is queued successfully and released by Netty, or fails
                    // to be queued.
                    // Otherwise, a memory leak will occur.
                    monos.add(userSession.sendNotification(notificationData, tracingContext)
                            .onErrorResume(t -> {
                                offlineRecipientIds.add(recipientId);
                                if (userSession.isSessionOpen()) {
                                    return Mono.error(new RuntimeException(
                                            "Failed to send a notification to the user session: "
                                                    + userSession));
                                }
                                return Mono.empty();
                            }));
                    userSession.getConnection()
                            .tryNotifyClientToRecover();
                }
            }
        }
        return Mono.deferContextual(context -> Mono.whenDelayError(monos)
                .doOnEach(signal -> {
                    boolean isOnError = signal.isOnError();
                    if (!signal.isOnComplete() && !isOnError) {
                        return;
                    }
                    if (pluginManager.hasRunningExtensions(NotificationHandler.class)) {
                        invokeExtensionPointHandlers(notificationData,
                                recipientIds,
                                offlineRecipientIds);
                    }
                    SimpleTurmsNotification notification = isNotificationLoggingEnabled
                            ? TurmsNotificationParser.parseSimpleNotification(
                                    ProtoDecoder.newInputStream(notificationData))
                            : null;
                    Throwable t = signal.getThrowable();
                    if (isNotificationLoggingEnabled
                            && apiLoggingContext
                                    .shouldLogNotification(notification.relayedRequestType())) {
                        int offlineRecipientCount = offlineRecipientIds.size();
                        int notificationBytes = notificationData.readableBytes();
                        try (TracingCloseableContext ignored =
                                TracingContext.getCloseableContext(context)) {
                            notificationLoggingManager.log(notification,
                                    notificationBytes,
                                    recipientCount,
                                    recipientCount - offlineRecipientCount);
                            if (t != null) {
                                LOGGER.error(
                                        "Caught an error while sending a notification to user sessions",
                                        t);
                            }
                        }
                    } else if (t != null) {
                        try (TracingCloseableContext ignored =
                                TracingContext.getCloseableContext(context)) {
                            LOGGER.error(
                                    "Caught an error while sending a notification to user sessions",
                                    t);
                        }
                    }
                })
                .onErrorComplete(t -> true)
                .thenReturn(offlineRecipientIds)
                .doFinally(signalType -> notificationData.release()));
    }
@param notificationData should be a buffer of TurmsNotification, and it is always a
                        "notification" instead of "response" in fact
@return offline recipient IDs
@implNote 1. The method ensures notificationData will be released by 1 2. No need to
          updateMdc for trace ID here because we know sendNotificationToLocalClients() is in
          the tracing scope
turms-gateway/src/main/java/im/turms/gateway/domain/notification/service/NotificationService.java