diff --git a/.github/workflows/ci-urma.yml b/.github/workflows/ci-urma.yml
index 4b8fe52bee..174ddb9a2c 100644
--- a/.github/workflows/ci-urma.yml
+++ b/.github/workflows/ci-urma.yml
@@ -66,7 +66,8 @@ jobs:
         run: |
           targets=(brpc-static brpc_urma_unittest brpc_urma_recovery_unittest
             brpc_urma_runtime_unittest brpc_urma_copy_unittest
-            brpc_urma_transport_unittest brpc_urma_remote_recovery_unittest)
+            brpc_urma_transport_unittest brpc_urma_remote_recovery_unittest
+            brpc_urma_mock_completion_unittest)
           if test -f test/brpc_urma_qos_unittest.cpp; then
             targets+=(brpc_urma_qos_unittest brpc_urma_qos_churn_unittest)
           fi
diff --git a/docs/cn/urma.md b/docs/cn/urma.md
index 68342cdaf3..d90447e5a6 100644
--- a/docs/cn/urma.md
+++ b/docs/cn/urma.md
@@ -256,3 +256,46 @@ URMA 恢复及 64 KiB 附件回显。正常短连接关闭也在同一测试中
 `--urma_copy_send=true` 将每个数据 WR 合并到 64 字节对齐的注册缓冲区，并持有至完成，
 增加一次 CPU 拷贝。两端保持设备、队列和消息大小一致，对比原发送路径与复制路径。
 初始化日志记录 `copy_send`，失败日志记录完成方向、槽位及注册段范围。
+
+## 业务 SL 与优先级
+
+`ChannelOptions.urma_qos` 控制请求发送方向，`ServerOptions.urma_qos` 控制响应发送方向：
+
+```cpp
+brpc::ChannelOptions client_options;
+client_options.socket_mode = brpc::SOCKET_MODE_URMA;
+client_options.urma_qos.service_level = 2;
+
+brpc::ServerOptions server_options;
+server_options.socket_mode = brpc::SOCKET_MODE_URMA;
+server_options.urma_qos.service_level = 2;
+```
+
+`--urma_tp_type=auto` 按设备能力优先选择 CTP，其次 RTP；`ctp`、`rtp` 显式选择路径。
+两端使用相同路径，进程初始化后保持固定。设备未提供能力表时，auto/ctp 使用 priority 6，
+rtp 使用 priority 0。
+
+| 配置 | 默认 | 显式范围 | 含义 |
+| --- | --- | --- | --- |
+| `service_level` | -1 | 0–255 | 选择当前路径下首个映射到该 SL 的 priority |
+| `priority` | -1 | 0–15 | 指定当前路径支持的 priority |
+
+两项均为 -1 时使用设备默认 priority；同时指定时须匹配同一设备映射。
+不支持或冲突时，Channel 初始化、Server 启动报错，并列出当前路径支持的 priority:SL。
+例如 RTP priority 0→SL 2 时配置 `service_level=2`，Jetty 使用 priority 0。
+平台配置设备映射与调度策略，brpc 使用进程初始化时的映射快照。
+
+不同配置使用独立连接，相同配置共享连接。配置随服务发现、连接池、短连接和
+服务端 accept 传递，Jetty 重建时重新应用。预建池服务于默认 priority，
+其他 priority 按需创建。v2/v3 握手使用所选路径。
+
+### 功能验证
+
+用 `urma_admin show --dev "$URMA_DEVICE" --whole` 核对当前路径的 priority:SL。
+`example/urma_performance` 的两端均支持 `--urma_tp_type=rtp --urma_sl=2`，
+也可通过 `--urma_priority` 指定匹配项。先以默认配置完成 RPC，再分别改变请求和响应 SL，
+验证合法映射、不支持的 SL 和 SL/priority 冲突。对 v2/v3、single/pooled 及 polling 模式重复验证。
+
+连接调试信息提供 `state`、`urma_effective_priority` 和 `urma_effective_sl`。
+核对 URMA 连接为 `ESTABLISHED`、映射符合设备配置、RPC 无错误。
+调度效果测试结合竞争流量、设备队列计数器、吞吐与 P99。
diff --git a/docs/en/urma.md b/docs/en/urma.md
index f3fd8139cf..7c1c999686 100644
--- a/docs/en/urma.md
+++ b/docs/en/urma.md
@@ -279,3 +279,54 @@ including normal short-connection closes before the injection.
 buffer held until completion, adding one CPU copy. Keep device, queue and payload
 settings consistent when comparing send paths. Startup logs report `copy_send`;
 failure logs report direction, send slot and registered segment range.
+
+## Business SL and priority
+
+Set `ChannelOptions.urma_qos` for outgoing requests and
+`ServerOptions.urma_qos` for outgoing responses:
+
+```cpp
+brpc::ChannelOptions client_options;
+client_options.socket_mode = brpc::SOCKET_MODE_URMA;
+client_options.urma_qos.service_level = 2;
+
+brpc::ServerOptions server_options;
+server_options.socket_mode = brpc::SOCKET_MODE_URMA;
+server_options.urma_qos.service_level = 2;
+```
+
+`--urma_tp_type=auto` prefers CTP, then RTP, using device capabilities.
+Use `ctp` or `rtp` to select a path explicitly. Both peers use the same path,
+fixed at process initialization. Providers without a capability table use
+priority 6 for auto/CTP and priority 0 for RTP.
+
+| Option | Default | Explicit range | Meaning |
+| --- | --- | --- | --- |
+| `service_level` | -1 | 0–255 | Select the first priority mapping to this SL on the selected path |
+| `priority` | -1 | 0–15 | Select a priority supported by the selected path |
+
+Both fields at -1 select the device default. When both are explicit, they must
+match the same device mapping. Channel initialization and Server startup reject
+unsupported or conflicting settings and report supported priority:SL pairs.
+For example, RTP priority 0 mapping to SL 2 resolves `service_level=2` to priority 0.
+The platform configures mappings and scheduling; brpc uses the initialization snapshot.
+
+Different configurations use separate connections; equal configurations share
+connections. Naming services, pooled and short connections, accepted sockets and
+rebuilt Jettys inherit the configuration. The prepared pool serves the default
+priority; other priorities allocate Jettys on demand. Both v2/v3 handshakes carry
+the selected path.
+
+### Functional validation
+
+Inspect the selected path's priority:SL map with
+`urma_admin show --dev "$URMA_DEVICE" --whole`.
+Both examples in `example/urma_performance` accept `--urma_tp_type=rtp --urma_sl=2`
+and an optional matching `--urma_priority`. Establish RPCs with defaults, then
+change request and response SLs separately. Check supported mappings, unsupported
+SLs and conflicting pairs across v2/v3, single/pooled and polling modes.
+
+Connection diagnostics expose `state`, `urma_effective_priority` and
+`urma_effective_sl`. Check that URMA is `ESTABLISHED`, the mapping matches the
+device and RPCs succeed. Evaluate scheduling with competing traffic, device queue
+counters, throughput and P99.
diff --git a/example/urma_performance/client.cpp b/example/urma_performance/client.cpp
index 9449dbe960..8f7e29570a 100644
--- a/example/urma_performance/client.cpp
+++ b/example/urma_performance/client.cpp
@@ -44,6 +44,9 @@ DEFINE_int32(attachment_size, -1, "Attachment size is used (in Bytes)");
 DEFINE_int32(rpc_timeout_ms, 5000, "Timeout for each RPC in milliseconds");
 DEFINE_bool(echo_attachment, false, "Select whether attachment should be echo");
 DEFINE_bool(use_urma, true, "Use URMA transport (true) or TCP (false)");
+DEFINE_int32(urma_priority, -1,
+             "Outbound URMA priority: -1 automatic, or [0,15]");
+DEFINE_int32(urma_sl, -1, "Outbound URMA SL: -1 unspecified, or [0,255]");
 
 bvar::LatencyRecorder g_latency("client");
 bvar::Adder<int64_t> g_error_count("client_error_count");
@@ -88,6 +91,8 @@ static void* worker(void* arg) {
 int main(int argc, char* argv[]) {
     gflags::ParseCommandLineFlags(&argc, &argv, true);
     brpc::ChannelOptions options;
+    options.urma_qos.priority = FLAGS_urma_priority;
+    options.urma_qos.service_level = FLAGS_urma_sl;
     options.socket_mode = FLAGS_use_urma ? brpc::SOCKET_MODE_URMA
                                           : brpc::SOCKET_MODE_TCP;
     options.connect_timeout_ms = FLAGS_rpc_timeout_ms;
@@ -110,7 +115,9 @@ int main(int argc, char* argv[]) {
     stub.Test(&warmup_cntl, &warmup_req, &warmup_resp, nullptr);
     if (warmup_cntl.Failed()) {
         LOG(ERROR) << "Warm-up RPC failed after timeout_ms="
-                   << FLAGS_rpc_timeout_ms << ": "
+                   << FLAGS_rpc_timeout_ms << " use_urma=" << FLAGS_use_urma
+                   << " urma_priority=" << FLAGS_urma_priority
+                   << " urma_sl=" << FLAGS_urma_sl << ": "
                    << warmup_cntl.ErrorText();
         return -1;
     }
@@ -131,7 +138,8 @@ int main(int argc, char* argv[]) {
     }
     LOG(INFO) << "URMA performance client started (server=" << FLAGS_server
               << ", use_urma=" << FLAGS_use_urma
-              << ", threads=" << thread_num
+              << ", urma_priority=" << FLAGS_urma_priority
+              << ", urma_sl=" << FLAGS_urma_sl << ", threads=" << thread_num
               << ", rpc_timeout_ms=" << FLAGS_rpc_timeout_ms << ")";
     while (!brpc::IsAskedToQuit()) {
         sleep(1);
diff --git a/example/urma_performance/server.cpp b/example/urma_performance/server.cpp
index baaff323b6..e5f8649487 100644
--- a/example/urma_performance/server.cpp
+++ b/example/urma_performance/server.cpp
@@ -30,6 +30,9 @@
 
 DEFINE_int32(port, 8003, "TCP Port of this server");
 DEFINE_bool(use_urma, true, "Use URMA transport (true) or TCP (false)");
+DEFINE_int32(urma_priority, -1,
+             "Outbound URMA priority: -1 automatic, or [0,15]");
+DEFINE_int32(urma_sl, -1, "Outbound URMA SL: -1 unspecified, or [0,255]");
 
 butil::atomic<uint64_t> g_last_time(0);
 
@@ -73,6 +76,8 @@ int main(int argc, char* argv[]) {
     }
 
     brpc::ServerOptions options;
+    options.urma_qos.priority = FLAGS_urma_priority;
+    options.urma_qos.service_level = FLAGS_urma_sl;
     options.socket_mode = FLAGS_use_urma ? brpc::SOCKET_MODE_URMA
                                           : brpc::SOCKET_MODE_TCP;
     if (server.Start(FLAGS_port, &options) != 0) {
@@ -80,7 +85,9 @@ int main(int argc, char* argv[]) {
         return -1;
     }
     LOG(INFO) << "URMA performance server started on port " << FLAGS_port
-              << " (use_urma=" << FLAGS_use_urma << ")";
+              << " (use_urma=" << FLAGS_use_urma
+              << ", urma_priority=" << FLAGS_urma_priority
+              << ", urma_sl=" << FLAGS_urma_sl << ")";
     server.RunUntilAskedToQuit();
     return 0;
 }
diff --git a/src/brpc/acceptor.cpp b/src/brpc/acceptor.cpp
index 99280037d5..36a3f3e827 100644
--- a/src/brpc/acceptor.cpp
+++ b/src/brpc/acceptor.cpp
@@ -338,6 +338,7 @@ void Acceptor::OnNewConnectionsUntilEAGAIN(Socket* acception) {
         options.force_ssl = am->_force_ssl;
         options.initial_ssl_ctx = am->_ssl_ctx;
         options.socket_mode = am->_socket_mode;
+        options.urma_qos = am->_urma_qos;
         options.bthread_tag = am->_bthread_tag;
         if (Socket::Create(options, &socket_id) != 0) {
             am->ReleaseConnectionSlot();
diff --git a/src/brpc/acceptor.h b/src/brpc/acceptor.h
index 01a2b26a9d..661994f528 100644
--- a/src/brpc/acceptor.h
+++ b/src/brpc/acceptor.h
@@ -24,7 +24,7 @@
 #include "butil/containers/flat_map.h"
 #include "brpc/input_messenger.h"
 #include "brpc/socket_mode.h"
-
+#include "brpc/urma/urma_qos.h"
 
 namespace brpc {
 
@@ -137,6 +137,7 @@ friend class Server;
 
     // Choose to use a certain socket: 0 TCP, 1 RDMA
     SocketMode _socket_mode;
+    UrmaQosOptions _urma_qos;
 
     // Acceptor belongs to this tag
     bthread_tag_t _bthread_tag;
diff --git a/src/brpc/channel.cpp b/src/brpc/channel.cpp
index 7b5f0a0da6..4f3cf6ab85 100644
--- a/src/brpc/channel.cpp
+++ b/src/brpc/channel.cpp
@@ -78,11 +78,9 @@ ChannelSSLOptions* ChannelOptions::mutable_ssl_options() {
 }
 
 static ChannelSignature ComputeChannelSignature(const ChannelOptions& opt) {
-    if (opt.auth == nullptr &&
-        !opt.has_ssl_options() &&
-        opt.client_host.empty() &&
-        opt.device_name.empty() &&
-        opt.connection_group.empty() &&
+    if (opt.socket_mode != SOCKET_MODE_URMA && opt.auth == nullptr &&
+        !opt.has_ssl_options() && opt.client_host.empty() &&
+        opt.device_name.empty() && opt.connection_group.empty() &&
         opt.hc_option.health_check_path.empty()) {
         // Returning zeroized result by default is more intuitive for users.
         return ChannelSignature();
@@ -141,6 +139,10 @@ static ChannelSignature ComputeChannelSignature(const ChannelOptions& opt) {
             buf.append("|rdma");
         } else if (opt.socket_mode == SOCKET_MODE_URMA) {
             buf.append("|urma");
+            buf.append("|priority=")
+                .append(std::to_string(opt.urma_qos.priority));
+            buf.append("|sl=").append(
+                std::to_string(opt.urma_qos.service_level));
         }
         butil::MurmurHash3_x64_128_Update(&mm_ctx, buf.data(), buf.size());
         buf.clear();
@@ -187,6 +189,12 @@ int Channel::InitChannelOptions(const ChannelOptions* options) {
     if (options) {  // Override default options if user provided one.
         _options = *options;
     }
+    if (_options.socket_mode == SOCKET_MODE_URMA &&
+        !_options.urma_qos.IsValid()) {
+        LOG(ERROR) << "URMA priority must be -1 or in [0,15], "
+                      "and SL must be -1 or in [0,255]";
+        return -1;
+    }
     const Protocol* protocol = FindProtocol(_options.protocol);
     if (nullptr == protocol || !protocol->support_client()) {
         LOG(ERROR) << "Channel does not support the protocol";
@@ -407,6 +415,7 @@ int Channel::InitSingle(const butil::EndPoint& server_addr_and_port,
     opt.local_side = client_endpoint;
     opt.initial_ssl_ctx = ssl_ctx;
     opt.socket_mode = _options.socket_mode;
+    opt.urma_qos = _options.urma_qos;
     opt.hc_option = _options.hc_option;
     opt.device_name = _options.device_name;
     if (SocketMapInsert(SocketMapKey(server_addr_and_port, sig),
@@ -453,6 +462,7 @@ int Channel::Init(const char* ns_url,
     ns_opt.succeed_without_server = _options.succeed_without_server;
     ns_opt.log_succeed_without_server = _options.log_succeed_without_server;
     ns_opt.socket_option.socket_mode = _options.socket_mode;
+    ns_opt.socket_option.urma_qos = _options.urma_qos;
     ns_opt.channel_signature = ComputeChannelSignature(_options);
     ns_opt.socket_option.hc_option =  _options.hc_option;
     ns_opt.socket_option.local_side = client_endpoint;
diff --git a/src/brpc/channel.h b/src/brpc/channel.h
index a778a1bd2a..e295b5275f 100644
--- a/src/brpc/channel.h
+++ b/src/brpc/channel.h
@@ -38,6 +38,7 @@
 #include "brpc/naming_service_filter.h"
 #include "brpc/health_check_option.h"
 #include "brpc/socket_mode.h"
+#include "brpc/urma/urma_qos.h"
 
 namespace brpc {
 
@@ -111,6 +112,9 @@ struct ChannelOptions {
     // Default: SOCKET_MODE_TCP
     SocketMode socket_mode;
 
+    // Outbound URMA QoS for this Channel. Used when socket_mode is URMA.
+    UrmaQosOptions urma_qos;
+
     // Turn on authentication for this channel if `auth' is not nullptr.
     // Note `auth' will not be deleted by channel and must remain valid when
     // the channel is being used.
diff --git a/src/brpc/server.cpp b/src/brpc/server.cpp
index 3dcacb96f8..1a70543d18 100644
--- a/src/brpc/server.cpp
+++ b/src/brpc/server.cpp
@@ -876,6 +876,13 @@ int Server::StartInternal(const butil::EndPoint& endpoint,
     const ServerOptions default_opt;
     const ServerOptions& real_opt = opt ? *opt : default_opt;
 
+    if (real_opt.socket_mode == SOCKET_MODE_URMA &&
+        !real_opt.urma_qos.IsValid()) {
+        LOG(ERROR) << "URMA priority must be -1 or in [0,15], "
+                      "and SL must be -1 or in [0,255]";
+        return -1;
+    }
+
     if (!real_opt.h2_settings.IsValid(true/*log_error*/)) {
         LOG(ERROR) << "Invalid h2_settings";
         return -1;
@@ -1163,6 +1170,7 @@ int Server::StartInternal(const butil::EndPoint& endpoint,
             _am->_socket_mode = _options.socket_mode;
             _am->_bthread_tag = _options.bthread_tag;
         }
+        _am->_urma_qos = _options.urma_qos;
         // Set `_status' to RUNNING before accepting connections
         // to prevent requests being rejected as ELOGOFF
         _status = RUNNING;
diff --git a/src/brpc/server.h b/src/brpc/server.h
index 31dc69af2f..130d181a9c 100644
--- a/src/brpc/server.h
+++ b/src/brpc/server.h
@@ -46,6 +46,7 @@
 #include "brpc/baidu_master_service.h"
 #include "brpc/rpc_pb_message_factory.h"
 #include "brpc/socket_mode.h"
+#include "brpc/urma/urma_qos.h"
 
 namespace brpc {
 
@@ -242,6 +243,9 @@ struct ServerOptions {
     // Default: SOCKET_MODE_TCP
     SocketMode socket_mode;
 
+    // Outbound URMA QoS for responses on accepted connections.
+    UrmaQosOptions urma_qos;
+
     // [CAUTION] This option is for implementing specialized baidu-std proxies,
     // most users don't need it. Don't change this option unless you fully
     // understand the description below.
diff --git a/src/brpc/socket.cpp b/src/brpc/socket.cpp
index c4411c9606..856fb9ec48 100644
--- a/src/brpc/socket.cpp
+++ b/src/brpc/socket.cpp
@@ -711,6 +711,10 @@ void SetSocketBufferOptions(int fd) {
 //   version: from version part of _versioned_nref, must be an EVEN number.
 //   slot: designated by ResourcePool.
 int Socket::Create(const SocketOptions& options, SocketId* id) {
+    if (options.socket_mode == SOCKET_MODE_URMA && !options.urma_qos.IsValid()) {
+        errno = EINVAL;
+        return -1;
+    }
     return VersionedRefWithId<Socket>::Create(id, options);
 }
 
@@ -726,6 +730,7 @@ int Socket::OnCreated(const SocketOptions& options) {
     });
     // start build the transport
     _socket_mode = options.socket_mode;
+    _urma_qos = options.urma_qos;
     _transport = TransportFactory::CreateTransport(options.socket_mode);
     CHECK(nullptr != _transport);
     _transport->Init(this, options);
@@ -2814,6 +2819,7 @@ SocketOptions Socket::GetSubSocketOptions() const {
     opt.keytable_pool = _keytable_pool;
     opt.app_connect = _app_connect;
     opt.socket_mode = _socket_mode;
+    opt.urma_qos = _urma_qos;
     opt.keepalive_options = _keepalive_options;
     opt.tcp_user_timeout_ms = _tcp_user_timeout_ms;
     opt.bthread_tag = _io_event.bthread_tag();
diff --git a/src/brpc/socket.h b/src/brpc/socket.h
index 8a1d372294..a17eef3265 100644
--- a/src/brpc/socket.h
+++ b/src/brpc/socket.h
@@ -44,6 +44,7 @@
 #include "brpc/versioned_ref_with_id.h"
 #include "brpc/health_check_option.h"
 #include "brpc/socket_mode.h"
+#include "brpc/urma/urma_qos.h"
 #include "brpc/input_messenger_processor.h"  // InputMessengerProcessor
 
 namespace brpc {
@@ -310,6 +311,7 @@ struct SocketOptions {
     bool force_ssl{false};
     std::shared_ptr<SocketSSLContext> initial_ssl_ctx;
     SocketMode socket_mode{SOCKET_MODE_TCP};
+    UrmaQosOptions urma_qos;
     bthread_keytable_pool_t* keytable_pool{nullptr};
     SocketConnection* conn{nullptr};
     std::shared_ptr<AppConnect> app_connect;
@@ -640,6 +642,7 @@ friend class TransportFactory;
 
     // The socket mode decided at creation.
     SocketMode socket_mode() const { return _socket_mode; }
+    const UrmaQosOptions& urma_qos() const { return _urma_qos; }
 
     // Get an UNUSED socket connecting to the same place as this socket
     // from the SocketPool of this socket.
@@ -1017,6 +1020,7 @@ friend class TransportFactory;
 
     // Should use SOCKET_MODE_RDMA or SOCKET_MODE_TCP or Other, default is SOCKET_MODE_TCP Transport
     SocketMode _socket_mode;
+    UrmaQosOptions _urma_qos;
     std::unique_ptr<Transport> _transport;
 
     // Pass from controller, for progressive reading.
diff --git a/src/brpc/urma/mock_urma.cpp b/src/brpc/urma/mock_urma.cpp
index f6b89db1ec..51e1e4d089 100644
--- a/src/brpc/urma/mock_urma.cpp
+++ b/src/brpc/urma/mock_urma.cpp
@@ -240,14 +240,20 @@ urma_status_t urma_query_device(urma_device_t *device,
     if (!device || !attr) {
         return URMA_EINVAL;
     }
+    mock_device_attr.dev_cap.priority_info[3].tp_type.bs.rtp = 1;
+    mock_device_attr.dev_cap.priority_info[3].SL = 3;
+    mock_device_attr.dev_cap.priority_info[6].tp_type.bs.ctp = 1;
+    mock_device_attr.dev_cap.priority_info[6].tp_type.bs.rtp = 1;
+    mock_device_attr.dev_cap.priority_info[6].SL = 2;
+    mock_device_attr.dev_cap.priority_info[10].tp_type.bs.ctp = 1;
+    mock_device_attr.dev_cap.priority_info[10].tp_type.bs.rtp = 1;
+    mock_device_attr.dev_cap.priority_info[10].SL = 4;
     mock_device_attr.dev_cap.max_jfc = 1024;
     mock_device_attr.dev_cap.max_jetty = 1024;
     mock_device_attr.dev_cap.max_jfs_sge = 8;
     mock_device_attr.dev_cap.max_jfr_sge = 8;
     mock_device_attr.dev_cap.priority_info[0].tp_type.bs.rtp = 1;
     mock_device_attr.dev_cap.priority_info[0].SL = 2;
-    mock_device_attr.dev_cap.priority_info[6].tp_type.bs.ctp = 1;
-    mock_device_attr.dev_cap.priority_info[6].SL = 6;
     memcpy(attr, &mock_device_attr, sizeof(urma_device_attr_t));
     return URMA_SUCCESS;
 }
@@ -566,14 +572,16 @@ urma_status_t urma_post_jetty_send_wr(urma_jetty_t *jetty, urma_jfs_wr_t *wr,
             return URMA_EINVAL;
         }
 
-        urma_cr_t send_cr{};
-        send_cr.status = URMA_CR_SUCCESS;
-        send_cr.user_ctx = current->user_ctx;
-        send_cr.flag.bs.s_r = 0;
-        if (current->flag.bs.complete_enable) {
-            PushCompletion(jetty->jetty_cfg.jfs_cfg.jfc,
-                           local_state, send_cr);
-        }
+        auto complete_send = [&] {
+            if (current->flag.bs.complete_enable) {
+                urma_cr_t send_cr{};
+                send_cr.status = URMA_CR_SUCCESS;
+                send_cr.user_ctx = current->user_ctx;
+                send_cr.flag.bs.s_r = 0;
+                PushCompletion(jetty->jetty_cfg.jfs_cfg.jfc,
+                               local_state, send_cr);
+            }
+        };
 
         PendingRecv recv{};
         urma_jfc_t* remote_jfc = nullptr;
@@ -581,6 +589,7 @@ urma_status_t urma_post_jetty_send_wr(urma_jetty_t *jetty, urma_jfs_wr_t *wr,
             std::unique_lock<std::shared_mutex> lock(g_rw_mutex);
             auto remote_it = jetty_id_map.find(current->tjetty->id.id);
             if (remote_it == jetty_id_map.end()) {
+                complete_send();
                 continue;
             }
             urma_jfr_t* remote_jfr =
@@ -589,6 +598,7 @@ urma_status_t urma_post_jetty_send_wr(urma_jetty_t *jetty, urma_jfs_wr_t *wr,
             auto jfc_it = jfr_jfc_map.find(remote_jfr);
             if (recv_it == jfr_recv_map.end() || recv_it->second.empty() ||
                 jfc_it == jfr_jfc_map.end()) {
+                complete_send();
                 continue;
             }
             recv = recv_it->second.front();
@@ -627,6 +637,9 @@ urma_status_t urma_post_jetty_send_wr(urma_jetty_t *jetty, urma_jfs_wr_t *wr,
             recv_cr.imm_data = current->send.imm_data;
             PushCompletion(remote_jfc, remote_state, recv_cr);
         }
+        // A SEND completion lets the consumer release its source buffers.
+        // Finish copying the payload before publishing that completion.
+        complete_send();
     }
     if (bad_wr) {
         *bad_wr = nullptr;
diff --git a/src/brpc/urma/urma_endpoint.cpp b/src/brpc/urma/urma_endpoint.cpp
index d4119397fb..c35a69676c 100644
--- a/src/brpc/urma/urma_endpoint.cpp
+++ b/src/brpc/urma/urma_endpoint.cpp
@@ -394,6 +394,10 @@ int UrmaEndpoint::AllocateResources() {
     if (_resource) {
         return 0;
     }
+    uint8_t priority;
+    if (!ResolveUrmaJettyPriority(_socket->urma_qos(), &priority)) {
+        return -1;
+    }
     urma_context_t* ctx = GetUrmaContext();
     if (!ctx) {
         errno = ENODEV;
@@ -406,7 +410,8 @@ int UrmaEndpoint::AllocateResources() {
     }
 
     // Try the prepared pool first (sized sq/rq match).
-    if (_sq_size <= static_cast<uint16_t>(FLAGS_urma_sq_size) &&
+    if (priority == GetUrmaJettyPriority() &&
+        _sq_size <= static_cast<uint16_t>(FLAGS_urma_sq_size) &&
         _rq_size <= static_cast<uint16_t>(FLAGS_urma_rq_size)) {
         BAIDU_SCOPED_LOCK(g_prepared_mutex);
         if (g_prepared_list) {
@@ -455,7 +460,7 @@ int UrmaEndpoint::AllocateResources() {
         jetty_cfg.flag.bs.share_jfr = 1;
         jetty_cfg.jfs_cfg.depth = static_cast<uint32_t>(_sq_size);
         jetty_cfg.jfs_cfg.trans_mode = URMA_TM_RM;
-        jetty_cfg.jfs_cfg.priority = GetUrmaJettyPriority();
+        jetty_cfg.jfs_cfg.priority = priority;
         jetty_cfg.jfs_cfg.max_sge =
             static_cast<uint8_t>(GetUrmaMaxSge());
         jetty_cfg.jfs_cfg.rnr_retry = URMA_TYPICAL_RNR_RETRY;
@@ -970,6 +975,18 @@ ssize_t UrmaEndpoint::HandleCompletion(const urma_cr_t& cr) {
     if (cr.status != URMA_CR_SUCCESS) {
         LOG(WARNING) << "URMA completion failed, status=" << cr.status
                      << " direction=" << (cr.flag.bs.s_r ? "recv" : "send")
+                     << " urma_priority=" << _socket->urma_qos().priority
+                     << " urma_sl=" << _socket->urma_qos().service_level
+                     << " urma_effective_priority="
+                     << (_resource && _resource->jetty
+                             ? static_cast<int>(
+                                   _resource->jetty->jetty_cfg.jfs_cfg.priority)
+                             : -1)
+                     << " urma_effective_sl="
+                     << (_resource && _resource->jetty
+                             ? GetUrmaServiceLevelForPriority(
+                                   _resource->jetty->jetty_cfg.jfs_cfg.priority)
+                             : -1)
                      << " flags=" << static_cast<unsigned>(cr.flag.value)
                      << " opcode=" << cr.opcode << " local_id=" << cr.local_id
                      << " tpn=" << cr.tpn << " user_ctx=" << cr.user_ctx
@@ -1709,6 +1726,17 @@ void UrmaEndpoint::DebugInfo(std::ostream& os, butil::StringPiece) const {
        << " sq_window=" << _sq_window_size.load(butil::memory_order_relaxed)
        << " remote_rq_window="
        << _remote_rq_window_size.load(butil::memory_order_relaxed)
+       << " urma_priority=" << _socket->urma_qos().priority
+       << " urma_sl=" << _socket->urma_qos().service_level
+       << " urma_effective_priority="
+       << (_resource && _resource->jetty
+               ? static_cast<int>(_resource->jetty->jetty_cfg.jfs_cfg.priority)
+               : -1)
+       << " urma_effective_sl="
+       << (_resource && _resource->jetty
+               ? GetUrmaServiceLevelForPriority(
+                     _resource->jetty->jetty_cfg.jfs_cfg.priority)
+               : -1)
        << " recovery_epoch=" << _recovery_epoch
        << " runtime_failure=" << UrmaFailureReasonName(_runtime_failure.load())
        << " handshake_version=" << _handshake_version
diff --git a/src/brpc/urma/urma_endpoint.h b/src/brpc/urma/urma_endpoint.h
index 033f7c9d45..44366e90e3 100644
--- a/src/brpc/urma/urma_endpoint.h
+++ b/src/brpc/urma/urma_endpoint.h
@@ -103,6 +103,7 @@ struct UrmaResource {
 // false sharing between connections.
 class BAIDU_CACHELINE_ALIGNMENT UrmaEndpoint : public SocketUser {
     friend class UrmaRecoveryTestPeer;
+    friend class UrmaQosTestPeer;
     friend class UrmaConnect;
     friend class Socket;
     friend class UrmaTransport;
diff --git a/src/brpc/urma/urma_helper.cpp b/src/brpc/urma/urma_helper.cpp
index 47d9193f4c..3ce3b63b77 100644
--- a/src/brpc/urma/urma_helper.cpp
+++ b/src/brpc/urma/urma_helper.cpp
@@ -29,6 +29,7 @@
 #include <atomic>
 #include <functional>
 #include <limits>
+#include <sstream>
 #include <new>
 #include <utility>
 #include <vector>
@@ -712,7 +713,8 @@ static bool GlobalUrmaInitializeImpl() {
               << " bonding=" << g_is_bonding_device
               << " tp_type=" << (g_tp_type == URMA_CTP ? "ctp" : "rtp")
               << " priority=" << static_cast<unsigned>(g_jetty_priority)
-              << " max_sge=" << g_max_sge
+              << " max_sge=" << g_max_sge << " default_sl="
+              << GetUrmaServiceLevelForPriority(g_jetty_priority)
               << " copy_send=" << FLAGS_urma_copy_send
               << " buffer_size=" << g_pool_buffer_size
               << " buffer_count=" << g_pool->buffer_count();
@@ -804,6 +806,105 @@ bool SelectUrmaTransportPath(const urma_device_attr_t& attr,
 }
 urma_tp_type_t GetUrmaTpType() { return g_tp_type; }
 uint8_t GetUrmaJettyPriority() { return g_jetty_priority; }
+bool IsUrmaPrioritySupported(const urma_device_attr_t& attr, int priority,
+                             urma_tp_type_t tp_type) {
+    return priority >= 0 && priority <= URMA_MAX_PRIORITY &&
+           (tp_type == URMA_CTP || tp_type == URMA_RTP) &&
+           (attr.dev_cap.priority_info[priority].tp_type.value &
+            TpTypeCapability(tp_type).value) != 0;
+}
+
+static std::string DescribeUrmaSlMap(const urma_device_attr_t& attr,
+                                     urma_tp_type_t tp_type) {
+    std::ostringstream os;
+    os << '[';
+    bool first = true;
+    for (int priority = 0; priority <= URMA_MAX_PRIORITY; ++priority) {
+        if (IsUrmaPrioritySupported(attr, priority, tp_type)) {
+            if (!first) {
+                os << ',';
+            }
+            first = false;
+            os << priority << ':' << attr.dev_cap.priority_info[priority].SL;
+        }
+    }
+    os << ']';
+    return os.str();
+}
+
+bool ResolveUrmaJettyPriority(const urma_device_attr_t& attr,
+                              const UrmaQosOptions& qos, urma_tp_type_t tp_type,
+                              uint8_t default_priority, uint8_t* priority) {
+    if (!qos.IsValid() || !priority ||
+        (tp_type != URMA_CTP && tp_type != URMA_RTP)) {
+        errno = EINVAL;
+        return false;
+    }
+    if (qos.priority == -1 && qos.service_level == -1) {
+        *priority = default_priority;
+        return true;
+    }
+    if (qos.priority == -1) {
+        for (int p = 0; p <= URMA_MAX_PRIORITY; ++p) {
+            if (IsUrmaPrioritySupported(attr, p, tp_type) &&
+                attr.dev_cap.priority_info[p].SL ==
+                    static_cast<uint32_t>(qos.service_level)) {
+                *priority = static_cast<uint8_t>(p);
+                return true;
+            }
+        }
+        LOG(ERROR) << "URMA SL " << qos.service_level << " has no "
+                   << (tp_type == URMA_CTP ? "CTP" : "RTP")
+                   << " priority; supported priority:SL="
+                   << DescribeUrmaSlMap(attr, tp_type);
+    } else if (!IsUrmaPrioritySupported(attr, qos.priority, tp_type)) {
+        LOG(ERROR) << "URMA priority " << qos.priority << " does not advertise "
+                   << (tp_type == URMA_CTP ? "CTP" : "RTP")
+                   << " on the selected device; requested SL="
+                   << qos.service_level << "; supported priority:SL="
+                   << DescribeUrmaSlMap(attr, tp_type);
+    } else if (qos.service_level != -1 &&
+               attr.dev_cap.priority_info[qos.priority].SL !=
+                   static_cast<uint32_t>(qos.service_level)) {
+        LOG(ERROR) << "URMA priority " << qos.priority << " maps to SL "
+                   << attr.dev_cap.priority_info[qos.priority].SL
+                   << ", requested SL=" << qos.service_level
+                   << "; supported priority:SL="
+                   << DescribeUrmaSlMap(attr, tp_type);
+    } else {
+        *priority = static_cast<uint8_t>(qos.priority);
+        return true;
+    }
+    errno = EINVAL;
+    return false;
+}
+
+bool ResolveUrmaJettyPriority(const UrmaQosOptions& qos, uint8_t* priority) {
+    if (!priority) {
+        errno = EINVAL;
+        return false;
+    }
+    if (qos.IsValid() && qos.priority == -1 && qos.service_level == -1) {
+        *priority = GetUrmaJettyPriority();
+        return true;
+    }
+    if (!IsUrmaAvailable() || !g_context) {
+        LOG(ERROR) << "URMA QoS requires an initialized device: priority="
+                   << qos.priority << " SL=" << qos.service_level;
+        errno = EINVAL;
+        return false;
+    }
+    return ResolveUrmaJettyPriority(g_device_attr, qos, GetUrmaTpType(),
+                                    GetUrmaJettyPriority(), priority);
+}
+
+int64_t GetUrmaServiceLevelForPriority(int priority) {
+    if (!g_context ||
+        !IsUrmaPrioritySupported(g_device_attr, priority, GetUrmaTpType())) {
+        return -1;
+    }
+    return g_device_attr.dev_cap.priority_info[priority].SL;
+}
 int GetUrmaMaxSge() { return g_max_sge; }
 size_t GetUrmaRecvBlockSize() { return g_recv_block_size; }
 
diff --git a/src/brpc/urma/urma_helper.h b/src/brpc/urma/urma_helper.h
index 5cebf6cfdd..c5284cb651 100644
--- a/src/brpc/urma/urma_helper.h
+++ b/src/brpc/urma/urma_helper.h
@@ -33,6 +33,7 @@ class IOBuf;
 #if BRPC_WITH_URMA
 
 #include "urma_api.h"
+#include "brpc/urma/urma_qos.h"
 #include "urma_types.h"
 
 namespace brpc {
@@ -94,6 +95,19 @@ bool SelectUrmaTransportPath(const urma_device_attr_t& attr,
 urma_tp_type_t GetUrmaTpType();
 uint8_t GetUrmaJettyPriority();
 
+// Check whether a priority advertises the selected RM path.
+bool IsUrmaPrioritySupported(const urma_device_attr_t& attr, int priority,
+                             urma_tp_type_t tp_type);
+// Resolve a requested priority/SL pair against a device capability snapshot.
+// Fully automatic options retain default_priority, including legacy fallback.
+bool ResolveUrmaJettyPriority(const urma_device_attr_t& attr,
+                              const UrmaQosOptions& qos, urma_tp_type_t tp_type,
+                              uint8_t default_priority, uint8_t* priority);
+// Resolve automatic selection or validate against the initialized device.
+bool ResolveUrmaJettyPriority(const UrmaQosOptions& qos, uint8_t* priority);
+// Return the advertised SL on the selected path, or -1 when no initialized mapping is available.
+int64_t GetUrmaServiceLevelForPriority(int priority);
+
 // Initialize a local SGE whose complete byte range belongs to this context.
 bool InitUrmaLocalSge(urma_sge_t* sge, uint64_t address, size_t length,
                       urma_target_seg_t* segment);
diff --git a/src/brpc/urma/urma_qos.h b/src/brpc/urma/urma_qos.h
new file mode 100644
index 0000000000..56f5c43d28
--- /dev/null
+++ b/src/brpc/urma/urma_qos.h
@@ -0,0 +1,38 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#ifndef BRPC_URMA_QOS_H
+#define BRPC_URMA_QOS_H
+
+namespace brpc {
+
+// Outbound RM QoS, resolved through the selected device's priority/SL map.
+struct UrmaQosOptions {
+    // -1 selects from SL or the device default; explicit range [0, 15].
+    int priority = -1;
+    // -1 leaves SL unspecified; explicit range [0, 255].
+    int service_level = -1;
+
+    bool IsValid() const {
+        return priority >= -1 && priority <= 15 && service_level >= -1 &&
+               service_level <= 255;
+    }
+};
+
+}  // namespace brpc
+
+#endif  // BRPC_URMA_QOS_H
diff --git a/src/brpc/urma_transport.cpp b/src/brpc/urma_transport.cpp
index c5cc163ffa..fb3b8b2513 100644
--- a/src/brpc/urma_transport.cpp
+++ b/src/brpc/urma_transport.cpp
@@ -197,6 +197,12 @@ int UrmaTransport::ContextInitOrDie(bool server_or_not, const void* options) {
             return -1;
         }
         urma::GlobalUrmaInitializeOrDie();
+        uint8_t priority;
+        if (!urma::ResolveUrmaJettyPriority(
+                static_cast<const ServerOptions*>(options)->urma_qos,
+                &priority)) {
+            return -1;
+        }
         if (!urma::InitPollingModeWithTag(
                 static_cast<const ServerOptions*>(options)->bthread_tag)) {
             return -1;
@@ -207,6 +213,12 @@ int UrmaTransport::ContextInitOrDie(bool server_or_not, const void* options) {
             return -1;
         }
         urma::GlobalUrmaInitializeOrDie();
+        uint8_t priority;
+        if (!urma::ResolveUrmaJettyPriority(
+                static_cast<const ChannelOptions*>(options)->urma_qos,
+                &priority)) {
+            return -1;
+        }
         if (!urma::InitPollingModeWithTag(bthread_self_tag())) {
             return -1;
         }
diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt
index 60e98d496a..c83466f23d 100644
--- a/test/CMakeLists.txt
+++ b/test/CMakeLists.txt
@@ -360,6 +360,18 @@ foreach(BRPC_UT ${BRPC_UNITTESTS})
     else()
         add_test(NAME ${BRPC_UT_WE} COMMAND ${BRPC_UT_WE})
     endif()
+    if(WITH_URMA AND BRPC_UT_WE STREQUAL "brpc_urma_qos_unittest")
+        add_test(NAME brpc_urma_qos_rtp_unittest
+                 COMMAND ${BRPC_UT_WE} --urma_tp_type=rtp)
+        add_test(NAME brpc_urma_qos_rtp_polling_unittest
+                 COMMAND ${BRPC_UT_WE} --urma_tp_type=rtp --urma_use_polling=true)
+    endif()
+    if(WITH_URMA AND BRPC_UT_WE STREQUAL "brpc_urma_qos_churn_unittest")
+        add_test(NAME brpc_urma_qos_churn_rtp_unittest
+                 COMMAND ${BRPC_UT_WE} --urma_tp_type=rtp)
+        add_test(NAME brpc_urma_qos_churn_rtp_polling_unittest
+                 COMMAND ${BRPC_UT_WE} --urma_tp_type=rtp --urma_use_polling=true)
+    endif()
     if(WITH_URMA AND BRPC_UT_WE STREQUAL "brpc_urma_transport_unittest")
         add_test(NAME brpc_urma_transport_rtp_unittest
                  COMMAND ${BRPC_UT_WE} --urma_tp_type=rtp)
diff --git a/test/brpc_urma_mock_completion_unittest.cpp b/test/brpc_urma_mock_completion_unittest.cpp
new file mode 100644
index 0000000000..5a6173fb82
--- /dev/null
+++ b/test/brpc_urma_mock_completion_unittest.cpp
@@ -0,0 +1,162 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+
+#include <atomic>
+#include <chrono>
+#include <cstring>
+#include <string>
+#include <thread>
+
+#if BRPC_WITH_URMA
+#include "urma_api.h"
+
+namespace {
+
+class UrmaMockCompletionTest : public testing::Test {
+   protected:
+    void SetUp() override {
+        urma_init_attr_t attr{};
+        const auto status = urma_init(&attr);
+        ASSERT_TRUE(status == URMA_SUCCESS || status == URMA_EEXIST);
+        _owns_init = status == URMA_SUCCESS;
+        int count = 0;
+        auto** devices = urma_get_device_list(&count);
+        const bool mock =
+            devices && count > 0 &&
+            std::strcmp(devices[0]->name, "mock_urma_device") == 0;
+        if (mock) {
+            _context = urma_create_context(devices[0], 0);
+        }
+        urma_free_device_list(devices);
+        if (!mock) {
+            GTEST_SKIP() << "Completion ordering test uses the link-time mock";
+        }
+        ASSERT_NE(nullptr, _context);
+        urma_jfc_cfg_t cq{};
+        cq.depth = 8;
+        _send_cq = urma_create_jfc(_context, &cq);
+        _recv_cq = urma_create_jfc(_context, &cq);
+        ASSERT_NE(nullptr, _send_cq);
+        ASSERT_NE(nullptr, _recv_cq);
+        urma_jfr_cfg_t rq{};
+        rq.depth = 8;
+        rq.jfc = _recv_cq;
+        _recv_queue = urma_create_jfr(_context, &rq);
+        ASSERT_NE(nullptr, _recv_queue);
+        urma_jetty_cfg_t cfg{};
+        cfg.jfs_cfg.jfc = _send_cq;
+        cfg.shared.jfr = _recv_queue;
+        _sender = urma_create_jetty(_context, &cfg);
+        _receiver = urma_create_jetty(_context, &cfg);
+        ASSERT_NE(nullptr, _sender);
+        ASSERT_NE(nullptr, _receiver);
+        urma_rjetty_t remote{};
+        remote.jetty_id = _receiver->jetty_id;
+        urma_token_t token{};
+        _target = urma_import_jetty(_context, &remote, &token);
+        ASSERT_NE(nullptr, _target);
+    }
+
+    void TearDown() override {
+        if (_target) urma_unimport_jetty(_target);
+        if (_receiver) urma_delete_jetty(_receiver);
+        if (_sender) urma_delete_jetty(_sender);
+        if (_recv_queue) urma_delete_jfr(_recv_queue);
+        if (_recv_cq) urma_delete_jfc(_recv_cq);
+        if (_send_cq) urma_delete_jfc(_send_cq);
+        if (_context) urma_delete_context(_context);
+        if (_owns_init) urma_uninit();
+    }
+
+    bool _owns_init = false;
+    urma_context_t* _context = nullptr;
+    urma_jfc_t* _send_cq = nullptr;
+    urma_jfc_t* _recv_cq = nullptr;
+    urma_jfr_t* _recv_queue = nullptr;
+    urma_jetty_t* _sender = nullptr;
+    urma_jetty_t* _receiver = nullptr;
+    urma_target_jetty_t* _target = nullptr;
+};
+
+TEST_F(UrmaMockCompletionTest, send_completion_follows_payload_delivery) {
+    // Keep the copy large enough for a concurrent CQ consumer to run while
+    // the mock is still using the source buffer.
+    const std::string payload(16 * 1024 * 1024, 'p');
+    std::string received(payload.size(), '\0');
+    urma_sge_t recv_sge{};
+    recv_sge.addr = reinterpret_cast<uint64_t>(&received[0]);
+    recv_sge.len = received.size();
+    urma_jfr_wr_t recv_wr{};
+    recv_wr.src.sge = &recv_sge;
+    recv_wr.src.num_sge = 1;
+    recv_wr.user_ctx = 2;
+    urma_jfr_wr_t* bad_recv = nullptr;
+    ASSERT_EQ(URMA_SUCCESS, urma_post_jfr_wr(_recv_queue, &recv_wr, &bad_recv));
+
+    urma_sge_t send_sge{};
+    send_sge.addr = reinterpret_cast<uint64_t>(payload.data());
+    send_sge.len = payload.size();
+    urma_jfs_wr_t send_wr{};
+    send_wr.opcode = URMA_OPC_SEND;
+    send_wr.flag.bs.complete_enable = 1;
+    send_wr.tjetty = _target;
+    send_wr.send.src.sge = &send_sge;
+    send_wr.send.src.num_sge = 1;
+    send_wr.user_ctx = 1;
+    std::atomic<bool> started{false};
+    std::atomic<bool> finished{false};
+    urma_status_t posted = URMA_EINVAL;
+    std::thread writer([&] {
+        while (!started.load(std::memory_order_acquire)) {
+            std::this_thread::yield();
+        }
+        urma_jfs_wr_t* bad_send = nullptr;
+        posted = urma_post_jetty_send_wr(_sender, &send_wr, &bad_send);
+        finished.store(true, std::memory_order_release);
+    });
+    started.store(true, std::memory_order_release);
+    urma_cr_t send_cr{};
+    int send_count = 0;
+    const auto deadline =
+        std::chrono::steady_clock::now() + std::chrono::seconds(5);
+    while (send_count == 0 && std::chrono::steady_clock::now() < deadline) {
+        send_count = urma_poll_jfc(_send_cq, 1, &send_cr);
+        if (finished.load(std::memory_order_acquire) && send_count == 0) {
+            send_count = urma_poll_jfc(_send_cq, 1, &send_cr);
+            break;
+        }
+    }
+    urma_cr_t recv_cr{};
+    const int recv_count = urma_poll_jfc(_recv_cq, 1, &recv_cr);
+    // Join before assertions so a failed check still releases the worker.
+    writer.join();
+    ASSERT_EQ(URMA_SUCCESS, posted);
+    ASSERT_EQ(1, send_count);
+    EXPECT_EQ(URMA_CR_SUCCESS, send_cr.status);
+    EXPECT_EQ(1u, send_cr.user_ctx);
+    ASSERT_EQ(1, recv_count)
+        << "Mock SEND completed before delivering its source payload";
+    EXPECT_EQ(URMA_CR_SUCCESS, recv_cr.status);
+    EXPECT_EQ(2u, recv_cr.user_ctx);
+    EXPECT_EQ(payload.size(), recv_cr.completion_len);
+    EXPECT_EQ(payload, received);
+}
+
+}  // namespace
+#endif  // BRPC_WITH_URMA
diff --git a/test/brpc_urma_qos_churn_unittest.cpp b/test/brpc_urma_qos_churn_unittest.cpp
new file mode 100644
index 0000000000..2bc6d96ee9
--- /dev/null
+++ b/test/brpc_urma_qos_churn_unittest.cpp
@@ -0,0 +1,274 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <cstring>
+#include <functional>
+#include <string>
+#include <vector>
+#include <gflags/gflags.h>
+#include <gtest/gtest.h>
+
+#if BRPC_WITH_URMA
+#include "bthread/bthread.h"
+#include "butil/time.h"
+#include "brpc/acceptor.h"
+#include "brpc/channel.h"
+#include "brpc/closure_guard.h"
+#include "brpc/controller.h"
+#include "brpc/server.h"
+#include "brpc/urma/urma_endpoint.h"
+#include "brpc/urma/urma_helper.h"
+#include "brpc/urma/urma_recovery_policy.h"
+#include "brpc/urma_transport.h"
+#include "echo.pb.h"
+
+namespace brpc {
+namespace urma {
+DECLARE_int32(urma_buffer_count);
+DECLARE_int32(urma_prepared_jetty_cnt);
+DECLARE_int32(urma_client_handshake_version);
+extern bool g_skip_urma_init;
+
+class UrmaQosTestPeer {
+public:
+    static int Priority(UrmaEndpoint* endpoint) {
+        return endpoint->_resource && endpoint->_resource->jetty
+                   ? endpoint->_resource->jetty->jetty_cfg.jfs_cfg.priority
+                   : -1;
+    }
+    static ssize_t FailCompletion(UrmaEndpoint* endpoint) {
+        urma_cr_t cr{};
+        cr.status = URMA_CR_RNR_RETRY_CNT_EXC_ERR;
+        return endpoint->HandleCompletion(cr);
+    }
+};
+}  // namespace urma
+}  // namespace brpc
+
+namespace {
+using namespace brpc;
+
+bool WaitUntil(const std::function<bool()>& ready) {
+    const int64_t deadline = butil::monotonic_time_us() + 5000000;
+    while (!ready()) {
+        if (butil::monotonic_time_us() >= deadline) {
+            return false;
+        }
+        bthread_usleep(1000);
+    }
+    return true;
+}
+
+class QosEcho : public test::EchoService {
+    void Echo(google::protobuf::RpcController* base,
+              const test::EchoRequest* request, test::EchoResponse* response,
+              google::protobuf::Closure* done) override {
+        ClosureGuard guard(done);
+        auto* controller = static_cast<Controller*>(base);
+        response->set_message(request->message());
+        controller->response_attachment().append(
+            controller->request_attachment());
+    }
+};
+
+class UrmaQosChurnTest : public testing::TestWithParam<int> {
+protected:
+    static void SetUpTestSuite() {
+        urma_init_attr_t attr{};
+        const auto status = urma_init(&attr);
+        if (status != URMA_SUCCESS && status != URMA_EEXIST) {
+            return;
+        }
+        int count = 0;
+        auto** devices = urma_get_device_list(&count);
+        _mock = devices && count > 0 &&
+                std::strcmp(devices[0]->name, "mock_urma_device") == 0;
+        urma_free_device_list(devices);
+        if (status == URMA_SUCCESS) {
+            urma_uninit();
+        }
+        if (_mock) {
+            urma::g_skip_urma_init = false;
+            urma::FLAGS_urma_prepared_jetty_cnt = 2;
+            urma::GlobalUrmaInitializeOrDie();
+        }
+    }
+
+    void SetUp() override {
+        if (!_mock) {
+            GTEST_SKIP() << "URMA churn tests use the link-time mock";
+        }
+        ASSERT_TRUE(urma::IsUrmaAvailable());
+        _saved_mode = urma::GetUrmaRecoveryMode();
+        urma::SetUrmaRecoveryMode(urma::UrmaRecoveryMode::LEGACY);
+        urma::SetUrmaRecoveryMode(urma::UrmaRecoveryMode::AUTO);
+        ASSERT_TRUE(urma::SetUrmaRecoveryCooldownMs(1000));
+        urma::FLAGS_urma_client_handshake_version = GetParam();
+        ASSERT_EQ(0, _server.AddService(&_service, SERVER_DOESNT_OWN_SERVICE));
+        ServerOptions options;
+        options.socket_mode = SOCKET_MODE_URMA;
+        options.urma_qos.service_level = 2;
+        ASSERT_EQ(0, _server.Start("127.0.0.1:0", &options));
+        _options.socket_mode = SOCKET_MODE_URMA;
+        _options.urma_qos.service_level = 4;
+        _options.timeout_ms = 2000;
+        _options.connect_timeout_ms = 2000;
+        _options.max_retry = 0;
+        _options.connection_type = "single";
+    }
+
+    void TearDown() override {
+        if (_mock) {
+            _server.Stop(0);
+            _server.Join();
+            EXPECT_TRUE(WaitUntil([this] {
+                for (auto id : _client_ids) {
+                    SocketUniquePtr socket;
+                    if (Socket::AddressFailedAsWell(id, &socket) >= 0) {
+                        return false;
+                    }
+                }
+                return true;
+            })) << "Client sockets retained after test completion";
+            urma::SetUrmaRecoveryMode(urma::UrmaRecoveryMode::LEGACY);
+            urma::SetUrmaRecoveryMode(_saved_mode);
+            urma::SetUrmaRecoveryCooldownMs(5000);
+        }
+    }
+
+    void Echo(Channel* channel, int sequence) {
+        Controller controller;
+        test::EchoRequest request;
+        test::EchoResponse response;
+        request.set_message("QoS lifecycle " + std::to_string(sequence));
+        std::string payload(65536, '\0');
+        for (size_t i = 0; i < payload.size(); ++i) {
+            payload[i] = static_cast<char>((i + sequence) % 251);
+        }
+        controller.request_attachment().append(payload);
+        test::EchoService_Stub(channel).Echo(&controller, &request, &response,
+                                             nullptr);
+        ASSERT_FALSE(controller.Failed()) << controller.ErrorText();
+        EXPECT_EQ(request.message(), response.message());
+        EXPECT_EQ(payload, controller.response_attachment().to_string());
+    }
+
+    void CheckConnection(Channel* channel, bool use_urma) {
+        SocketUniquePtr socket;
+        ASSERT_EQ(0, Socket::Address(channel->_server_id, &socket));
+        EXPECT_EQ(4, socket->urma_qos().service_level);
+        EXPECT_EQ(-1, socket->urma_qos().priority);
+        auto* transport = static_cast<UrmaTransport*>(socket->_transport.get());
+        EXPECT_EQ(use_urma ? UrmaTransport::URMA_ON : UrmaTransport::URMA_OFF,
+                  transport->_urma_state.load());
+        EXPECT_EQ(use_urma ? 10 : -1,
+                  urma::UrmaQosTestPeer::Priority(transport->GetUrmaEp()));
+        std::vector<SocketId> accepted;
+        _server._am->ListConnections(&accepted);
+        bool found = false;
+        for (auto id : accepted) {
+            SocketUniquePtr peer;
+            if (Socket::Address(id, &peer) == 0 &&
+                peer->remote_side() == socket->local_side()) {
+                found = true;
+                EXPECT_EQ(2, peer->urma_qos().service_level);
+                auto* peer_transport =
+                    static_cast<UrmaTransport*>(peer->_transport.get());
+                const int expected_priority =
+                    use_urma ? (urma::GetUrmaTpType() == URMA_RTP ? 0 : 6) : -1;
+                EXPECT_EQ(expected_priority, urma::UrmaQosTestPeer::Priority(
+                                                 peer_transport->GetUrmaEp()));
+            }
+        }
+        EXPECT_TRUE(found);
+    }
+
+    static bool _mock;
+    GFLAGS_NAMESPACE::FlagSaver _flags;
+    urma::UrmaRecoveryMode _saved_mode = urma::UrmaRecoveryMode::LEGACY;
+    QosEcho _service;
+    Server _server;
+    ChannelOptions _options;
+    std::vector<SocketId> _client_ids;
+};
+bool UrmaQosChurnTest::_mock = false;
+
+TEST_P(UrmaQosChurnTest, short_connections_release_resources) {
+    _options.connection_type = "short";
+    Channel channel;
+    ASSERT_EQ(0, channel.Init(_server.listen_address(), &_options));
+    _client_ids.push_back(channel._server_id);
+    for (int i = 0; i < 64; ++i) {
+        ASSERT_NO_FATAL_FAILURE(Echo(&channel, i));
+        ASSERT_TRUE(WaitUntil([this] {
+            return _server._am->ConnectionCount() == 0;
+        })) << "Accepted short connection retained after RPC "
+            << i;
+    }
+}
+
+TEST_P(UrmaQosChurnTest, cooldown_and_fresh_connections_preserve_business_qos) {
+    Channel original;
+    ASSERT_EQ(0, original.Init(_server.listen_address(), &_options));
+    _client_ids.push_back(original._server_id);
+    ASSERT_NO_FATAL_FAILURE(Echo(&original, 1));
+    ASSERT_NO_FATAL_FAILURE(CheckConnection(&original, true));
+    const auto baseline = urma::GetUrmaRecoveryStatus();
+    SocketUniquePtr socket;
+    ASSERT_EQ(0, Socket::Address(original._server_id, &socket));
+    auto* transport = static_cast<UrmaTransport*>(socket->_transport.get());
+    EXPECT_EQ(-1,
+              urma::UrmaQosTestPeer::FailCompletion(transport->GetUrmaEp()));
+    socket->SetFailed(EIO, "Injected completion error");
+    socket.reset();
+
+    _options.connection_group = "qos-cooldown";
+    Channel tcp;
+    ASSERT_EQ(0, tcp.Init(_server.listen_address(), &_options));
+    _client_ids.push_back(tcp._server_id);
+    ASSERT_NO_FATAL_FAILURE(Echo(&tcp, 2));
+    ASSERT_NO_FATAL_FAILURE(CheckConnection(&tcp, false));
+    EXPECT_GT(urma::GetUrmaRecoveryStatus().peer_cooldown_admissions,
+              baseline.peer_cooldown_admissions);
+    ASSERT_TRUE(WaitUntil(
+        [] { return urma::GetUrmaRecoveryStatus().cooled_peers == 0; }));
+
+    _options.connection_group = "qos-recovered";
+    Channel recovered;
+    ASSERT_EQ(0, recovered.Init(_server.listen_address(), &_options));
+    _client_ids.push_back(recovered._server_id);
+    ASSERT_NO_FATAL_FAILURE(Echo(&recovered, 3));
+    ASSERT_NO_FATAL_FAILURE(CheckConnection(&recovered, true));
+    const auto after = urma::GetUrmaRecoveryStatus();
+    const size_t reason = static_cast<size_t>(urma::UrmaFailureReason::RNR);
+    EXPECT_EQ(baseline.failures[reason] + 1, after.failures[reason]);
+    EXPECT_GT(after.negotiated_connections, baseline.negotiated_connections);
+}
+
+INSTANTIATE_TEST_SUITE_P(HandshakeVersions, UrmaQosChurnTest,
+                         testing::Values(2, 3));
+}  // namespace
+#endif  // BRPC_WITH_URMA
+
+int main(int argc, char** argv) {
+    testing::InitGoogleTest(&argc, argv);
+#if BRPC_WITH_URMA
+    brpc::urma::FLAGS_urma_buffer_count = 4096;
+#endif
+    GFLAGS_NAMESPACE::ParseCommandLineFlags(&argc, &argv, true);
+    return RUN_ALL_TESTS();
+}
diff --git a/test/brpc_urma_qos_unittest.cpp b/test/brpc_urma_qos_unittest.cpp
new file mode 100644
index 0000000000..3a8e89c851
--- /dev/null
+++ b/test/brpc_urma_qos_unittest.cpp
@@ -0,0 +1,799 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <cstring>
+#include <sstream>
+#include <vector>
+#include <gtest/gtest.h>
+#include <gflags/gflags.h>
+
+#if BRPC_WITH_URMA
+#include "butil/logging.h"
+#include "butil/memory/scope_guard.h"
+#include "brpc/global.h"
+#include "brpc/acceptor.h"
+#include "brpc/channel.h"
+#include "brpc/controller.h"
+#include "brpc/server.h"
+#include "brpc/input_messenger.h"
+#include "brpc/load_balancer.h"
+#include "brpc/urma/urma_endpoint.h"
+#include "brpc/urma/urma_helper.h"
+#include "brpc/urma/urma_handshake.h"
+#include "brpc/urma_transport.h"
+#include "echo.pb.h"
+
+using namespace brpc;
+
+namespace brpc {
+namespace urma {
+DECLARE_int32(urma_client_handshake_version);
+DECLARE_string(urma_tp_type);
+extern bool g_skip_urma_init;
+DECLARE_int32(urma_buffer_count);
+DECLARE_int32(urma_prepared_jetty_cnt);
+
+class UrmaQosTestPeer {
+public:
+    static int Allocate(UrmaEndpoint* ep) { return ep->AllocateResources(); }
+    static int Priority(UrmaEndpoint* ep) {
+        return ep->_resource->jetty->jetty_cfg.jfs_cfg.priority;
+    }
+    static ssize_t Complete(UrmaEndpoint* ep, const urma_cr_t& cr) {
+        return ep->HandleCompletion(cr);
+    }
+    static int Import(UrmaEndpoint* ep, const ParsedHello& hello) {
+        return ep->ImportPeer(hello);
+    }
+    static ParsedHello Hello(UrmaEndpoint* ep) {
+        ParsedHello hello;
+        ep->MakeLocalParsedHello(&hello);
+        return hello;
+    }
+};
+
+}  // namespace urma
+}  // namespace brpc
+
+TEST(UrmaHelperTest, explicit_priority_requires_ctp_capability) {
+    urma_device_attr_t attr{};
+    attr.dev_cap.priority_info[0].tp_type.bs.ctp = 1;
+    attr.dev_cap.priority_info[15].tp_type.bs.ctp = 1;
+    attr.dev_cap.priority_info[15].tp_type.bs.rtp = 1;
+    attr.dev_cap.priority_info[3].tp_type.bs.rtp = 1;
+    EXPECT_TRUE(urma::IsUrmaPrioritySupported(attr, 0, URMA_CTP));
+    EXPECT_TRUE(urma::IsUrmaPrioritySupported(attr, 15, URMA_CTP));
+    for (int priority : {-2, -1, 3, 4, 16}) {
+        EXPECT_FALSE(urma::IsUrmaPrioritySupported(attr, priority, URMA_CTP));
+    }
+}
+
+TEST(UrmaHelperTest, automatic_priority_accepts_combined_capabilities) {
+    urma_device_attr_t attr{};
+    attr.dev_cap.priority_info[3].tp_type.bs.rtp = 1;
+    attr.dev_cap.priority_info[10].tp_type.bs.rtp = 1;
+    attr.dev_cap.priority_info[10].tp_type.bs.ctp = 1;
+    attr.dev_cap.priority_info[15].tp_type.bs.ctp = 1;
+    EXPECT_EQ(3, urma::FindUrmaPriorityForTpType(attr, URMA_RTP));
+    EXPECT_EQ(10, urma::FindUrmaPriorityForTpType(attr, URMA_CTP));
+    EXPECT_TRUE(urma::IsUrmaPrioritySupported(attr, 10, URMA_CTP));
+    EXPECT_EQ(-1, urma::FindUrmaPriorityForTpType(attr, URMA_UTP));
+    EXPECT_EQ(-1, urma::FindUrmaPriorityForTpType(
+                      attr, static_cast<urma_tp_type_t>(-1)));
+    attr.dev_cap.priority_info[10].tp_type.bs.ctp = 0;
+    EXPECT_EQ(15, urma::FindUrmaPriorityForTpType(attr, URMA_CTP));
+    attr.dev_cap.priority_info[15].tp_type.bs.ctp = 0;
+    EXPECT_EQ(-1, urma::FindUrmaPriorityForTpType(attr, URMA_CTP));
+}
+
+TEST(UrmaHelperTest, service_level_selects_ctp_priority_from_mapping) {
+    urma_device_attr_t attr{};
+    attr.dev_cap.priority_info[3].tp_type.bs.rtp = 1;
+    attr.dev_cap.priority_info[3].SL = 6;
+    attr.dev_cap.priority_info[6].tp_type.bs.ctp = 1;
+    attr.dev_cap.priority_info[6].SL = 7;
+    attr.dev_cap.priority_info[10].tp_type.bs.ctp = 1;
+    attr.dev_cap.priority_info[10].tp_type.bs.rtp = 1;
+    attr.dev_cap.priority_info[10].SL = 6;
+    attr.dev_cap.priority_info[15].tp_type.bs.ctp = 1;
+    attr.dev_cap.priority_info[15].SL = 6;
+    UrmaQosOptions qos;
+    uint8_t priority = 0;
+    qos.service_level = 6;
+    ASSERT_TRUE(
+        urma::ResolveUrmaJettyPriority(attr, qos, URMA_CTP, 6, &priority));
+    EXPECT_EQ(10, priority);
+    qos.service_level = 7;
+    ASSERT_TRUE(
+        urma::ResolveUrmaJettyPriority(attr, qos, URMA_CTP, 6, &priority));
+    EXPECT_EQ(6, priority);
+    qos.priority = 6;
+    ASSERT_TRUE(
+        urma::ResolveUrmaJettyPriority(attr, qos, URMA_CTP, 6, &priority));
+    EXPECT_EQ(6, priority);
+    attr.dev_cap.priority_info[15].SL = 255;
+    qos.priority = -1;
+    qos.service_level = 255;
+    ASSERT_TRUE(
+        urma::ResolveUrmaJettyPriority(attr, qos, URMA_CTP, 6, &priority));
+    EXPECT_EQ(15, priority);
+    qos.service_level = -1;
+    attr = urma_device_attr_t{};
+    ASSERT_TRUE(
+        urma::ResolveUrmaJettyPriority(attr, qos, URMA_CTP, 6, &priority));
+    EXPECT_EQ(6, priority);
+}
+
+TEST(UrmaHelperTest,
+     service_level_rejects_unsupported_and_conflicting_mapping) {
+    urma_device_attr_t attr{};
+    attr.dev_cap.priority_info[3].tp_type.bs.rtp = 1;
+    attr.dev_cap.priority_info[3].SL = 3;
+    attr.dev_cap.priority_info[6].tp_type.bs.ctp = 1;
+    attr.dev_cap.priority_info[6].SL = 2;
+    attr.dev_cap.priority_info[10].tp_type.bs.ctp = 1;
+    attr.dev_cap.priority_info[10].SL = 4;
+    UrmaQosOptions qos;
+    uint8_t priority = 255;
+    for (int sl : {-2, 0, 3, 6, 255, 256}) {
+        qos.service_level = sl;
+        EXPECT_FALSE(
+            urma::ResolveUrmaJettyPriority(attr, qos, URMA_CTP, 6, &priority));
+        EXPECT_EQ(EINVAL, errno);
+        EXPECT_EQ(255, priority);
+    }
+    qos.priority = 6;
+    qos.service_level = 4;
+    EXPECT_FALSE(
+        urma::ResolveUrmaJettyPriority(attr, qos, URMA_CTP, 6, &priority));
+    EXPECT_EQ(EINVAL, errno);
+    EXPECT_EQ(255, priority);
+#if !BRPC_WITH_GLOG
+    qos.priority = -1;
+    qos.service_level = 3;
+    logging::StringSink log;
+    logging::LogSink* old_sink = logging::SetLogSink(&log);
+    const bool resolved =
+        urma::ResolveUrmaJettyPriority(attr, qos, URMA_CTP, 6, &priority);
+    logging::SetLogSink(old_sink);
+    EXPECT_FALSE(resolved);
+    EXPECT_NE(std::string::npos, log.find("SL 3 has no CTP priority")) << log;
+    EXPECT_NE(std::string::npos, log.find("[6:2,10:4]")) << log;
+#endif
+}
+
+TEST(UrmaPathQosTest, automatic_and_explicit_paths_use_device_capabilities) {
+    urma_device_attr_t attr{};
+    attr.dev_cap.priority_info[0].tp_type.bs.rtp = 1;
+    attr.dev_cap.priority_info[0].SL = 2;
+    urma_tp_type_t path = URMA_CTP;
+    uint8_t priority = 15;
+    ASSERT_TRUE(urma::SelectUrmaTransportPath(attr, "auto", &path, &priority));
+    EXPECT_EQ(URMA_RTP, path);
+    EXPECT_EQ(0, priority);
+    attr.dev_cap.priority_info[6].tp_type.bs.ctp = 1;
+    attr.dev_cap.priority_info[6].tp_type.bs.rtp = 1;
+    ASSERT_TRUE(urma::SelectUrmaTransportPath(attr, "auto", &path, &priority));
+    EXPECT_EQ(URMA_CTP, path);
+    EXPECT_EQ(6, priority);
+    ASSERT_TRUE(urma::SelectUrmaTransportPath(attr, "rtp", &path, &priority));
+    EXPECT_EQ(URMA_RTP, path);
+    EXPECT_EQ(0, priority);
+}
+
+TEST(UrmaPathQosTest, unsupported_paths_preserve_outputs) {
+    urma_device_attr_t attr{};
+    attr.dev_cap.priority_info[0].tp_type.bs.rtp = 1;
+    urma_tp_type_t path = URMA_UTP;
+    uint8_t priority = 15;
+    for (const char* request : {"ctp", "utp", "", "typo"}) {
+        EXPECT_FALSE(
+            urma::SelectUrmaTransportPath(attr, request, &path, &priority));
+        EXPECT_EQ(URMA_UTP, path);
+        EXPECT_EQ(15, priority);
+    }
+    EXPECT_FALSE(
+        urma::SelectUrmaTransportPath(attr, "rtp", nullptr, &priority));
+    EXPECT_FALSE(urma::SelectUrmaTransportPath(attr, "rtp", &path, nullptr));
+    attr.dev_cap.priority_info[0].tp_type.value = 0;
+    attr.dev_cap.priority_info[4].tp_type.bs.utp = 1;
+    EXPECT_FALSE(urma::SelectUrmaTransportPath(attr, "auto", &path, &priority));
+}
+
+TEST(UrmaPathQosTest, empty_capability_table_uses_compatibility_defaults) {
+    urma_device_attr_t attr{};
+    urma_tp_type_t path = URMA_UTP;
+    uint8_t priority = 15;
+    for (const char* request : {"auto", "ctp", "rtp"}) {
+        ASSERT_TRUE(
+            urma::SelectUrmaTransportPath(attr, request, &path, &priority));
+        const bool rtp = std::string(request) == "rtp";
+        EXPECT_EQ(rtp ? URMA_RTP : URMA_CTP, path);
+        EXPECT_EQ(rtp ? 0 : 6, priority);
+        UrmaQosOptions qos;
+        uint8_t resolved = 15;
+        ASSERT_TRUE(urma::ResolveUrmaJettyPriority(attr, qos, path, priority,
+                                                   &resolved));
+        EXPECT_EQ(priority, resolved);
+        qos.service_level = 2;
+        EXPECT_FALSE(urma::ResolveUrmaJettyPriority(attr, qos, path, priority,
+                                                    &resolved));
+    }
+}
+
+TEST(UrmaPathQosTest, service_level_resolution_is_scoped_to_the_selected_path) {
+    urma_device_attr_t attr{};
+    attr.dev_cap.priority_info[0].tp_type.bs.rtp = 1;
+    attr.dev_cap.priority_info[0].SL = 2;
+    attr.dev_cap.priority_info[4].tp_type.bs.rtp = 1;
+    attr.dev_cap.priority_info[4].SL = 4;
+    attr.dev_cap.priority_info[6].tp_type.bs.ctp = 1;
+    attr.dev_cap.priority_info[6].SL = 2;
+    attr.dev_cap.priority_info[10].tp_type.bs.ctp = 1;
+    attr.dev_cap.priority_info[10].tp_type.bs.rtp = 1;
+    attr.dev_cap.priority_info[10].SL = 4;
+    UrmaQosOptions qos;
+    qos.service_level = 2;
+    uint8_t priority = 15;
+    ASSERT_TRUE(
+        urma::ResolveUrmaJettyPriority(attr, qos, URMA_RTP, 0, &priority));
+    EXPECT_EQ(0, priority);
+    ASSERT_TRUE(
+        urma::ResolveUrmaJettyPriority(attr, qos, URMA_CTP, 6, &priority));
+    EXPECT_EQ(6, priority);
+    qos.service_level = 4;
+    ASSERT_TRUE(
+        urma::ResolveUrmaJettyPriority(attr, qos, URMA_RTP, 0, &priority));
+    EXPECT_EQ(4, priority);
+    qos.priority = 10;
+    for (auto path : {URMA_RTP, URMA_CTP}) {
+        ASSERT_TRUE(
+            urma::ResolveUrmaJettyPriority(attr, qos, path, 0, &priority));
+        EXPECT_EQ(10, priority);
+    }
+}
+
+TEST(UrmaPathQosTest, rejects_cross_path_priorities_and_conflicting_sl) {
+    urma_device_attr_t attr{};
+    attr.dev_cap.priority_info[0].tp_type.bs.rtp = 1;
+    attr.dev_cap.priority_info[0].SL = 2;
+    attr.dev_cap.priority_info[6].tp_type.bs.ctp = 1;
+    attr.dev_cap.priority_info[6].SL = 6;
+    UrmaQosOptions qos;
+    uint8_t priority = 15;
+    qos.service_level = 6;
+    EXPECT_FALSE(
+        urma::ResolveUrmaJettyPriority(attr, qos, URMA_RTP, 0, &priority));
+    qos.priority = 6;
+    EXPECT_FALSE(
+        urma::ResolveUrmaJettyPriority(attr, qos, URMA_RTP, 0, &priority));
+    qos.priority = 0;
+    EXPECT_FALSE(
+        urma::ResolveUrmaJettyPriority(attr, qos, URMA_CTP, 6, &priority));
+    EXPECT_FALSE(
+        urma::ResolveUrmaJettyPriority(attr, qos, URMA_RTP, 0, &priority));
+    EXPECT_EQ(15, priority);
+    EXPECT_EQ(EINVAL, errno);
+}
+
+TEST(UrmaPathQosTest, rejects_invalid_resolution_arguments) {
+    urma_device_attr_t attr{};
+    UrmaQosOptions qos;
+    uint8_t priority = 15;
+    EXPECT_FALSE(
+        urma::ResolveUrmaJettyPriority(attr, qos, URMA_RTP, 0, nullptr));
+    EXPECT_FALSE(urma::ResolveUrmaJettyPriority(qos, nullptr));
+    for (auto path : {URMA_UTP, static_cast<urma_tp_type_t>(255)}) {
+        EXPECT_FALSE(
+            urma::ResolveUrmaJettyPriority(attr, qos, path, 0, &priority));
+        EXPECT_FALSE(urma::IsUrmaPrioritySupported(attr, 0, path));
+    }
+    EXPECT_EQ(15, priority);
+    EXPECT_EQ(EINVAL, errno);
+}
+
+class UrmaQosTest : public ::testing::Test {
+protected:
+    static void SetUpTestSuite() {
+        urma_init_attr_t attr{};
+        const auto status = urma_init(&attr);
+        if (status != URMA_SUCCESS && status != URMA_EEXIST) {
+            return;
+        }
+        int count = 0;
+        auto** devices = urma_get_device_list(&count);
+        _mock = devices && count > 0 &&
+                strcmp(devices[0]->name, "mock_urma_device") == 0;
+        urma_free_device_list(devices);
+        if (status == URMA_SUCCESS) {
+            urma_uninit();
+        }
+        if (!_mock) {
+            return;
+        }
+        urma::g_skip_urma_init = false;
+        urma::FLAGS_urma_buffer_count = 1024;
+        urma::FLAGS_urma_prepared_jetty_cnt = 2;
+        urma::GlobalUrmaInitializeOrDie();
+    }
+    void SetUp() override {
+        if (!_mock) {
+            GTEST_SKIP() << "URMA QoS integration tests use the link-time mock";
+        }
+        ASSERT_TRUE(urma::IsUrmaAvailable());
+        ASSERT_NE(nullptr, urma::GetUrmaContext());
+    }
+    static bool _mock;
+    static urma::UrmaEndpoint* Endpoint(SocketUniquePtr& socket) {
+        return static_cast<UrmaTransport*>(socket->_transport.get())
+            ->GetUrmaEp();
+    }
+    static int Sl2Priority() {
+        return urma::GetUrmaTpType() == URMA_RTP ? 0 : 6;
+    }
+};
+bool UrmaQosTest::_mock = false;
+
+TEST_F(UrmaQosTest, service_level_requires_capability_on_the_selected_path) {
+    ChannelOptions options;
+    options.socket_mode = SOCKET_MODE_URMA;
+    options.urma_qos.service_level = 3;
+    Channel channel;
+    if (urma::GetUrmaTpType() == URMA_CTP) {
+        EXPECT_EQ(-1, channel.Init("127.0.0.1:8123", &options));
+    } else {
+        ASSERT_EQ(0, channel.Init("127.0.0.1:8123", &options));
+        SocketUniquePtr socket;
+        ASSERT_EQ(0, Socket::Address(channel._server_id, &socket));
+        ASSERT_EQ(0, urma::UrmaQosTestPeer::Allocate(Endpoint(socket)));
+        EXPECT_EQ(3, urma::UrmaQosTestPeer::Priority(Endpoint(socket)));
+        EXPECT_EQ(3, urma::GetUrmaServiceLevelForPriority(3));
+    }
+}
+
+TEST_F(UrmaQosTest,
+       advertises_selected_path_and_rejects_mismatch_before_import) {
+    SocketOptions options;
+    options.socket_mode = SOCKET_MODE_URMA;
+    SocketId id;
+    ASSERT_EQ(0, Socket::Create(options, &id));
+    SocketUniquePtr socket;
+    ASSERT_EQ(0, Socket::Address(id, &socket));
+    auto* endpoint = Endpoint(socket);
+    auto hello = urma::UrmaQosTestPeer::Hello(endpoint);
+    EXPECT_EQ(static_cast<uint8_t>(urma::GetUrmaTpType()), hello.tp_type);
+    for (auto path :
+         {URMA_CTP, URMA_RTP, URMA_UTP, static_cast<urma_tp_type_t>(255)}) {
+        if (path == urma::GetUrmaTpType()) {
+            continue;
+        }
+        hello.tp_type = static_cast<uint8_t>(path);
+        EXPECT_EQ(-1, urma::UrmaQosTestPeer::Import(endpoint, hello));
+        EXPECT_EQ(EPROTO, errno);
+    }
+    socket->SetFailed();
+}
+
+TEST_F(UrmaQosTest, rejects_invalid_and_unsupported_priority) {
+    butil::EndPoint address;
+    ASSERT_EQ(0, butil::str2endpoint("127.0.0.1:8123", &address));
+    for (int priority : {-2, 16, 3, 4, 9}) {
+        if (priority == 3 && urma::GetUrmaTpType() == URMA_RTP) {
+            continue;
+        }
+        ChannelOptions co;
+        co.socket_mode = SOCKET_MODE_URMA;
+        co.urma_qos.priority = priority;
+        Channel channel;
+        EXPECT_EQ(-1, channel.Init(address, &co));
+        ServerOptions so;
+        so.socket_mode = SOCKET_MODE_URMA;
+        so.urma_qos = co.urma_qos;
+        Server server;
+        EXPECT_EQ(-1, server.Start("127.0.0.1:0", &so));
+        if (!co.urma_qos.IsValid()) {
+            SocketOptions options;
+            options.socket_mode = SOCKET_MODE_URMA;
+            options.urma_qos = co.urma_qos;
+            SocketId sid;
+            EXPECT_EQ(-1, Socket::Create(options, &sid));
+            EXPECT_EQ(EINVAL, errno);
+        } else {
+            SocketOptions options;
+            options.socket_mode = SOCKET_MODE_URMA;
+            options.urma_qos = co.urma_qos;
+            SocketId sid;
+            ASSERT_EQ(0, Socket::Create(options, &sid));
+            SocketUniquePtr socket;
+            ASSERT_EQ(0, Socket::Address(sid, &socket));
+            EXPECT_EQ(-1, urma::UrmaQosTestPeer::Allocate(Endpoint(socket)));
+            EXPECT_EQ(EINVAL, errno);
+            socket->SetFailed();
+        }
+    }
+}
+
+TEST_F(UrmaQosTest, rejects_invalid_unsupported_and_conflicting_service_level) {
+    struct RequestedQos {
+        int priority;
+        int sl;
+    };
+    for (const auto& requested : {RequestedQos{-1, -2},
+                                  {-1, 256},
+                                  {-1, 0},
+                                  {-1, 3},
+                                  {-1, 6},
+                                  {-1, 255},
+                                  {6, 4},
+                                  {3, 3},
+                                  {4, 3}}) {
+        if (requested.sl == 3 && requested.priority != 4 &&
+            urma::GetUrmaTpType() == URMA_RTP) {
+            continue;
+        }
+        ChannelOptions co;
+        co.socket_mode = SOCKET_MODE_URMA;
+        co.urma_qos.priority = requested.priority;
+        co.urma_qos.service_level = requested.sl;
+        Channel channel;
+        EXPECT_EQ(-1, channel.Init("127.0.0.1:8123", &co));
+        ServerOptions so;
+        so.socket_mode = SOCKET_MODE_URMA;
+        so.urma_qos = co.urma_qos;
+        Server server;
+        EXPECT_EQ(-1, server.Start("127.0.0.1:0", &so));
+        SocketOptions options;
+        options.socket_mode = SOCKET_MODE_URMA;
+        options.urma_qos = co.urma_qos;
+        SocketId sid;
+        if (!co.urma_qos.IsValid()) {
+            EXPECT_EQ(-1, Socket::Create(options, &sid));
+            EXPECT_EQ(EINVAL, errno);
+        } else {
+            ASSERT_EQ(0, Socket::Create(options, &sid));
+            SocketUniquePtr socket;
+            ASSERT_EQ(0, Socket::Address(sid, &socket));
+            EXPECT_EQ(-1, urma::UrmaQosTestPeer::Allocate(Endpoint(socket)));
+            EXPECT_EQ(EINVAL, errno);
+            socket->SetFailed();
+        }
+    }
+}
+
+TEST_F(UrmaQosTest,
+       channels_isolate_priority_and_inherit_to_child_connections) {
+    ChannelOptions co;
+    co.socket_mode = SOCKET_MODE_URMA;
+    Channel automatic, low, same, high, named, tcp, tcp_with_qos;
+    ASSERT_EQ(0, automatic.Init("127.0.0.1:8123", &co));
+    co.urma_qos.priority = 6;
+    ASSERT_EQ(0, low.Init("127.0.0.1:8123", &co));
+    ASSERT_EQ(0, same.Init("127.0.0.1:8123", &co));
+    ASSERT_EQ(0, named.Init("list://127.0.0.1:8123", "rr", &co));
+    co.urma_qos.priority = 10;
+    ASSERT_EQ(0, high.Init("127.0.0.1:8123", &co));
+    EXPECT_EQ(low._server_id, same._server_id);
+    EXPECT_NE(low._server_id, high._server_id);
+    EXPECT_NE(low._server_id, automatic._server_id);
+    co.socket_mode = SOCKET_MODE_TCP;
+    ASSERT_EQ(0, tcp_with_qos.Init("127.0.0.1:8123", &co));
+    ASSERT_EQ(0, tcp.Init("127.0.0.1:8123", nullptr));
+    EXPECT_EQ(tcp._server_id, tcp_with_qos._server_id);
+    EXPECT_NE(tcp._server_id, automatic._server_id);
+    SocketUniquePtr main, pooled, short_socket, selected;
+    ASSERT_EQ(0, Socket::Address(low._server_id, &main));
+    ASSERT_EQ(0, main->GetPooledSocket(&pooled));
+    ASSERT_EQ(0, main->GetShortSocket(&short_socket));
+    LoadBalancer::SelectIn input{};
+    LoadBalancer::SelectOut output(&selected);
+    ASSERT_EQ(0, named._lb->SelectServer(input, &output));
+    for (auto* socket :
+         {main.get(), pooled.get(), short_socket.get(), selected.get()}) {
+        EXPECT_EQ(6, socket->urma_qos().priority);
+    }
+    EXPECT_EQ(main->id(), selected->id());
+    EXPECT_EQ(0, pooled->ReturnToPool());
+    short_socket->SetFailed();
+}
+
+TEST_F(UrmaQosTest, sub_socket_qos_preserves_common_connection_options) {
+    brpc::GlobalInitializeOrDie();
+    SocketOptions options;
+    options.socket_mode = SOCKET_MODE_URMA;
+    options.urma_qos.priority = 10;
+    options.urma_qos.service_level = 4;
+    options.keepalive_options.emplace();
+    options.keepalive_options->keepalive_idle_s = 11;
+    options.keepalive_options->keepalive_interval_s = 12;
+    options.keepalive_options->keepalive_count = 13;
+    options.tcp_user_timeout_ms = 1400;
+    options.bthread_tag = bthread_self_tag() + 1;
+    SocketId id;
+    ASSERT_EQ(0, Socket::Create(options, &id));
+    BRPC_SCOPE_EXIT { Socket::SetFailed(id); };
+    SocketUniquePtr parent;
+    ASSERT_EQ(0, Socket::Address(id, &parent));
+    options.keepalive_options->keepalive_idle_s = 99;
+    for (int pooled = 0; pooled < 2; ++pooled) {
+        SCOPED_TRACE(pooled);
+        SocketUniquePtr child;
+        ASSERT_EQ(0, pooled ? parent->GetPooledSocket(&child)
+                            : parent->GetShortSocket(&child));
+        BRPC_SCOPE_EXIT { child->SetFailed(); };
+        EXPECT_EQ(SOCKET_MODE_URMA, child->socket_mode());
+        EXPECT_EQ(10, child->urma_qos().priority);
+        EXPECT_EQ(4, child->urma_qos().service_level);
+        ASSERT_TRUE(child->_keepalive_options.has_value());
+        EXPECT_EQ(11, child->_keepalive_options->keepalive_idle_s);
+        EXPECT_EQ(12, child->_keepalive_options->keepalive_interval_s);
+        EXPECT_EQ(13, child->_keepalive_options->keepalive_count);
+        EXPECT_EQ(1400, child->_tcp_user_timeout_ms);
+        EXPECT_EQ(options.bthread_tag, child->_io_event.bthread_tag());
+    }
+}
+
+TEST_F(UrmaQosTest, service_level_isolates_and_inherits_connections) {
+    ChannelOptions co;
+    co.socket_mode = SOCKET_MODE_URMA;
+    co.urma_qos.service_level = 2;
+    Channel low, same, high, named, by_priority, tcp, tcp_with_qos;
+    ASSERT_EQ(0, low.Init("127.0.0.1:8123", &co));
+    ASSERT_EQ(0, same.Init("127.0.0.1:8123", &co));
+    ASSERT_EQ(0, named.Init("list://127.0.0.1:8123", "rr", &co));
+    co.urma_qos.service_level = 4;
+    ASSERT_EQ(0, high.Init("127.0.0.1:8123", &co));
+    co.urma_qos.service_level = -1;
+    co.urma_qos.priority = 6;
+    ASSERT_EQ(0, by_priority.Init("127.0.0.1:8123", &co));
+    EXPECT_EQ(low._server_id, same._server_id);
+    EXPECT_NE(low._server_id, high._server_id);
+    EXPECT_NE(low._server_id, by_priority._server_id);
+    SocketUniquePtr main, pooled, short_socket, selected;
+    ASSERT_EQ(0, Socket::Address(low._server_id, &main));
+    ASSERT_EQ(0, main->GetPooledSocket(&pooled));
+    ASSERT_EQ(0, main->GetShortSocket(&short_socket));
+    LoadBalancer::SelectIn input{};
+    LoadBalancer::SelectOut output(&selected);
+    ASSERT_EQ(0, named._lb->SelectServer(input, &output));
+    for (auto* socket :
+         {main.get(), pooled.get(), short_socket.get(), selected.get()}) {
+        EXPECT_EQ(-1, socket->urma_qos().priority);
+        EXPECT_EQ(2, socket->urma_qos().service_level);
+    }
+    EXPECT_EQ(main->id(), selected->id());
+    EXPECT_EQ(0, pooled->ReturnToPool());
+    short_socket->SetFailed();
+    co.socket_mode = SOCKET_MODE_TCP;
+    co.urma_qos.service_level = 256;
+    ASSERT_EQ(0, tcp_with_qos.Init("127.0.0.1:8123", &co));
+    ASSERT_EQ(0, tcp.Init("127.0.0.1:8123", nullptr));
+    EXPECT_EQ(tcp._server_id, tcp_with_qos._server_id);
+}
+
+TEST_F(UrmaQosTest, prepared_jettys_are_used_only_for_matching_priority) {
+    urma::UrmaEndpoint::GlobalRelease();
+    ASSERT_EQ(0, urma::UrmaEndpoint::GlobalInitialize());
+    // SL 4 resolves to priority 10 and must leave the default Jettys available.
+    struct RequestedQos {
+        int priority;
+        int sl;
+        int effective;
+    };
+    for (const auto& qos : {RequestedQos{-1, 4, 10},
+                            {-1, 2, Sl2Priority()},
+                            {6, 2, 6},
+                            {10, -1, 10},
+                            {-1, -1, urma::GetUrmaJettyPriority()},
+                            {10, 4, 10}}) {
+        SocketOptions options;
+        options.socket_mode = SOCKET_MODE_URMA;
+        options.urma_qos.priority = qos.priority;
+        options.urma_qos.service_level = qos.sl;
+        SocketId sid;
+        ASSERT_EQ(0, Socket::Create(options, &sid));
+        SocketUniquePtr socket;
+        ASSERT_EQ(0, Socket::Address(sid, &socket));
+        auto* ep = Endpoint(socket);
+        ASSERT_EQ(0, urma::UrmaQosTestPeer::Allocate(ep));
+        EXPECT_EQ(qos.effective, urma::UrmaQosTestPeer::Priority(ep));
+        std::ostringstream debug;
+        ep->DebugInfo(debug);
+        EXPECT_NE(std::string::npos,
+                  debug.str().find("urma_effective_priority="));
+        EXPECT_NE(std::string::npos,
+                  debug.str().find("urma_sl=" + std::to_string(qos.sl)));
+        EXPECT_NE(std::string::npos,
+                  debug.str().find(
+                      "urma_effective_sl=" +
+                      std::to_string(urma::GetUrmaServiceLevelForPriority(
+                          qos.effective))));
+        socket->SetFailed();
+    }
+}
+
+TEST_F(UrmaQosTest, access_error_preserves_priority_and_reports_direction) {
+    for (int priority : {-1, 10}) {
+        SocketOptions options;
+        options.socket_mode = SOCKET_MODE_URMA;
+        options.urma_qos.priority = priority;
+        options.urma_qos.service_level = priority == -1 ? 2 : 4;
+        SocketId sid;
+        ASSERT_EQ(0, Socket::Create(options, &sid));
+        SocketUniquePtr socket;
+        ASSERT_EQ(0, Socket::Address(sid, &socket));
+        auto* ep = Endpoint(socket);
+        ASSERT_EQ(0, urma::UrmaQosTestPeer::Allocate(ep));
+        const int effective = priority == -1 ? Sl2Priority() : priority;
+        for (bool receive : {false, true}) {
+            urma_cr_t cr{};
+            cr.status = URMA_CR_LOC_ACCESS_ERR;
+            cr.flag.bs.s_r = receive;
+            cr.local_id = 42;
+#if !BRPC_WITH_GLOG
+            logging::StringSink log;
+            logging::LogSink* old_sink = logging::SetLogSink(&log);
+#endif
+            const ssize_t result = urma::UrmaQosTestPeer::Complete(ep, cr);
+            const int saved_errno = errno;
+#if !BRPC_WITH_GLOG
+            logging::SetLogSink(old_sink);
+            EXPECT_NE(std::string::npos, log.find("status=4")) << log;
+            EXPECT_NE(std::string::npos,
+                      log.find(receive ? "direction=recv" : "direction=send"))
+                << log;
+            EXPECT_NE(std::string::npos, log.find("local_id=42")) << log;
+            EXPECT_NE(std::string::npos,
+                      log.find("urma_priority=" + std::to_string(priority)))
+                << log;
+            EXPECT_NE(std::string::npos, log.find("urma_effective_priority=" +
+                                                  std::to_string(effective)))
+                << log;
+            EXPECT_NE(std::string::npos,
+                      log.find("urma_sl=" +
+                               std::to_string(options.urma_qos.service_level)))
+                << log;
+            EXPECT_NE(std::string::npos,
+                      log.find("urma_effective_sl=" +
+                               std::to_string(options.urma_qos.service_level)))
+                << log;
+#endif
+            EXPECT_EQ(-1, result);
+            EXPECT_EQ(EIO, saved_errno);
+            EXPECT_EQ(effective, urma::UrmaQosTestPeer::Priority(ep));
+        }
+        socket->SetFailed();
+    }
+}
+
+class UrmaQosEcho : public ::test::EchoService {
+    void Echo(google::protobuf::RpcController*,
+              const ::test::EchoRequest* request,
+              ::test::EchoResponse* response,
+              google::protobuf::Closure* done) override {
+        response->set_message(request->message());
+        done->Run();
+    }
+};
+
+TEST_F(UrmaQosTest, request_and_response_priorities_apply_across_restart) {
+    GFLAGS_NAMESPACE::FlagSaver saved_flags;
+    UrmaQosEcho service;
+    Server server;
+    ASSERT_EQ(0, server.AddService(&service, SERVER_DOESNT_OWN_SERVICE));
+    for (int response_priority : {6, 10}) {
+        urma::FLAGS_urma_client_handshake_version =
+            response_priority == 6 ? 2 : 3;
+        ServerOptions so;
+        so.socket_mode = SOCKET_MODE_URMA;
+        so.urma_qos.priority = response_priority;
+        ASSERT_EQ(0, server.Start("127.0.0.1:0", &so));
+        ChannelOptions co;
+        co.socket_mode = SOCKET_MODE_URMA;
+        co.urma_qos.priority = response_priority == 6 ? 10 : 6;
+        co.timeout_ms = 2000;
+        co.max_retry = 0;
+        co.connection_type = "single";
+        Channel channel;
+        ASSERT_EQ(0, channel.Init(server.listen_address(), &co));
+        Controller cntl;
+        ::test::EchoRequest request;
+        ::test::EchoResponse response;
+        request.set_message("URMA QoS");
+        ::test::EchoService_Stub(&channel).Echo(&cntl, &request, &response,
+                                                nullptr);
+        EXPECT_FALSE(cntl.Failed()) << cntl.ErrorText();
+        EXPECT_EQ(request.message(), response.message());
+        SocketUniquePtr client;
+        ASSERT_EQ(0, Socket::Address(channel._server_id, &client));
+        EXPECT_EQ(co.urma_qos.priority,
+                  urma::UrmaQosTestPeer::Priority(Endpoint(client)));
+        std::vector<SocketId> sockets;
+        server._am->ListConnections(&sockets);
+        ASSERT_EQ(1u, sockets.size());
+        SocketUniquePtr accepted;
+        ASSERT_EQ(0, Socket::Address(sockets[0], &accepted));
+        EXPECT_EQ(response_priority, accepted->urma_qos().priority);
+        EXPECT_EQ(response_priority,
+                  urma::UrmaQosTestPeer::Priority(Endpoint(accepted)));
+        accepted.reset();
+        client.reset();
+        server.Stop(0);
+        server.Join();
+    }
+}
+
+TEST_F(UrmaQosTest, request_and_response_service_levels_apply_across_restart) {
+    GFLAGS_NAMESPACE::FlagSaver saved_flags;
+    UrmaQosEcho service;
+    Server server;
+    ASSERT_EQ(0, server.AddService(&service, SERVER_DOESNT_OWN_SERVICE));
+    for (int response_sl : {2, 4}) {
+        urma::FLAGS_urma_client_handshake_version = response_sl == 2 ? 2 : 3;
+        ServerOptions so;
+        so.socket_mode = SOCKET_MODE_URMA;
+        so.urma_qos.service_level = response_sl;
+        ASSERT_EQ(0, server.Start("127.0.0.1:0", &so));
+        ChannelOptions co;
+        co.socket_mode = SOCKET_MODE_URMA;
+        co.urma_qos.service_level = response_sl == 2 ? 4 : 2;
+        co.timeout_ms = 2000;
+        co.max_retry = 0;
+        co.connection_type = "single";
+        Channel channel;
+        ASSERT_EQ(0, channel.Init(server.listen_address(), &co));
+        Controller cntl;
+        ::test::EchoRequest request;
+        ::test::EchoResponse response;
+        request.set_message("URMA SL mapping");
+        ::test::EchoService_Stub(&channel).Echo(&cntl, &request, &response,
+                                                nullptr);
+        EXPECT_FALSE(cntl.Failed()) << cntl.ErrorText();
+        EXPECT_EQ(request.message(), response.message());
+        SocketUniquePtr client;
+        ASSERT_EQ(0, Socket::Address(channel._server_id, &client));
+        EXPECT_EQ(-1, client->urma_qos().priority);
+        EXPECT_EQ(co.urma_qos.service_level, client->urma_qos().service_level);
+        EXPECT_EQ(response_sl == 2 ? 10 : Sl2Priority(),
+                  urma::UrmaQosTestPeer::Priority(Endpoint(client)));
+        std::vector<SocketId> sockets;
+        server._am->ListConnections(&sockets);
+        ASSERT_EQ(1u, sockets.size());
+        SocketUniquePtr accepted;
+        ASSERT_EQ(0, Socket::Address(sockets[0], &accepted));
+        EXPECT_EQ(-1, accepted->urma_qos().priority);
+        EXPECT_EQ(response_sl, accepted->urma_qos().service_level);
+        EXPECT_EQ(response_sl == 2 ? Sl2Priority() : 10,
+                  urma::UrmaQosTestPeer::Priority(Endpoint(accepted)));
+        std::ostringstream debug;
+        Endpoint(accepted)->DebugInfo(debug);
+        EXPECT_NE(std::string::npos,
+                  debug.str().find("urma_effective_sl=" +
+                                   std::to_string(response_sl)));
+        accepted.reset();
+        client.reset();
+        server.Stop(0);
+        server.Join();
+    }
+}
+
+#endif  // BRPC_WITH_URMA
+
+int main(int argc, char** argv) {
+    testing::InitGoogleTest(&argc, argv);
+    gflags::ParseCommandLineFlags(&argc, &argv, true);
+    return RUN_ALL_TESTS();
+}