private void invokeExtensionPointHandlers(
            @NotNull ByteBuf notificationData,
            @NotNull Set<Long> recipientIds,
            @NotNull Set<Long> offlineRecipientIds) {
        TurmsNotification notification;
        try {
            // Note that "parseFrom" won't block because the buffer is fully read
            notification =
                    TurmsNotification.parseFrom(ProtoDecoder.newInputStream(notificationData));
        } catch (Exception e) {
            LOGGER.error(
                    "Failed to parse TurmsNotification. All recipients' ID: {}. Offline recipients' ID: {}",
                    recipientIds,
                    offlineRecipientIds,
                    e);
            return;
        }
        pluginManager
                .invokeExtensionPointsSimultaneously(NotificationHandler.class,
                        HANDLE_METHOD,
                        handler -> handler.handle(notification, recipientIds, offlineRecipientIds))
                .subscribe(null, LOGGER::error);
    }
@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