diff --git a/docs/cn/rdma.md b/docs/cn/rdma.md
index 7dd2689156..cbe5e9934a 100644
--- a/docs/cn/rdma.md
+++ b/docs/cn/rdma.md
@@ -1,3 +1,24 @@
+# 业务级 QoS
+
+请求方向配置 `ChannelOptions.rdma_qos`，响应方向配置 `ServerOptions.rdma_qos`：
+
+```cpp
+brpc::ChannelOptions options;
+options.socket_mode = brpc::SOCKET_MODE_RDMA;
+options.rdma_qos.service_level = 3;
+options.rdma_qos.traffic_class = (26 << 2) | 2;  // DSCP=26, ECN=2
+```
+
+`service_level` 范围为 0–15，`traffic_class` 是完整的 8 位 GRH/IP 字节，范围为 0–255，
+两者默认均为 0。网络配置决定 SL→VL 及 PCP/DSCP 映射和调度策略。
+
+不同 QoS 使用独立连接与 QP，相同配置共享连接。配置随服务发现、连接池、短连接和
+服务端 accept 传递，在 QP 重建时重新应用。v2/v3 握手沿用现有格式。
+连接调试信息提供 `rdma_sl` 和 `rdma_traffic_class`。
+
+`example/rdma_performance` 的两端支持 `--rdma_sl`、`--rdma_traffic_class`。
+实机检查 QP 属性与报文标记，再结合竞争流量比较吞吐和 P99。
+
 # 运行期故障恢复
 
 模式切换、端口与对端冷却及 TCP 回落见
diff --git a/docs/cn/urma.md b/docs/cn/urma.md
index 48c822c622..9c9a848e12 100644
--- a/docs/cn/urma.md
+++ b/docs/cn/urma.md
@@ -249,3 +249,46 @@ python3 example/urma_performance/run_recovery_injection.py \
 `--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/rdma.md b/docs/en/rdma.md
index 98ac6981bc..19bbe5d0e6 100644
--- a/docs/en/rdma.md
+++ b/docs/en/rdma.md
@@ -1,3 +1,19 @@
+# Business QoS
+
+Configure `ChannelOptions.rdma_qos` for outgoing requests and
+`ServerOptions.rdma_qos` for outgoing responses. `service_level` accepts 0–15;
+`traffic_class` accepts the full 8-bit GRH/IP byte, 0–255, including DSCP and ECN.
+Both default to 0. The network configures SL→VL and PCP/DSCP mappings and scheduling.
+
+Different QoS settings use separate connections and QPs; equal settings share them.
+Naming services, pooled and short connections, accepted sockets and rebuilt QPs
+inherit the settings. v2/v3 handshake formats stay the same. Connection diagnostics
+show `rdma_sl` and `rdma_traffic_class`.
+
+Both examples in `example/rdma_performance` accept `--rdma_sl` and
+`--rdma_traffic_class`. Check QP attributes and packet markings, then compare
+throughput and P99 under competing traffic.
+
 # Build
 
 Since RDMA requires driver and hardware support, only the build on linux is verified.
diff --git a/docs/en/urma.md b/docs/en/urma.md
index 03cfc19098..2825753bb4 100644
--- a/docs/en/urma.md
+++ b/docs/en/urma.md
@@ -270,3 +270,54 @@ directory for the tool self-tests.
 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/rdma_performance/client.cpp b/example/rdma_performance/client.cpp
index 8c07eca236..365c61697d 100644
--- a/example/rdma_performance/client.cpp
+++ b/example/rdma_performance/client.cpp
@@ -42,6 +42,8 @@ DEFINE_string(connection_type, "single", "Connection type of the channel");
 DEFINE_string(protocol, "baidu_std", "Protocol type.");
 DEFINE_string(servers, "0.0.0.0:8002+0.0.0.0:8002", "IP Address of servers");
 DEFINE_bool(use_rdma, true, "Use RDMA or not");
+DEFINE_int32(rdma_sl, 0, "Outbound RDMA service level [0,15]");
+DEFINE_int32(rdma_traffic_class, 0, "Outbound RDMA traffic class byte [0,255]");
 DEFINE_int32(rpc_timeout_ms, 2000, "RPC call timeout");
 DEFINE_int32(test_seconds, 20, "Test running time");
 DEFINE_int32(test_iterations, 0, "Test iterations");
@@ -103,6 +105,8 @@ class PerformanceTest {
     int Init() {
         brpc::ChannelOptions options;
         options.socket_mode = FLAGS_use_rdma? brpc::SOCKET_MODE_RDMA : brpc::SOCKET_MODE_TCP;
+        options.rdma_qos.service_level = FLAGS_rdma_sl;
+        options.rdma_qos.traffic_class = FLAGS_rdma_traffic_class;
         options.protocol = FLAGS_protocol;
         options.connection_type = FLAGS_connection_type;
         options.timeout_ms = FLAGS_rpc_timeout_ms;
@@ -274,6 +278,14 @@ void Test(int thread_num, int attachment_size) {
 int main(int argc, char* argv[]) {
     GFLAGS_NAMESPACE::ParseCommandLineFlags(&argc, &argv, true);
 
+    brpc::RdmaQosOptions qos;
+    qos.service_level = FLAGS_rdma_sl;
+    qos.traffic_class = FLAGS_rdma_traffic_class;
+    if (FLAGS_use_rdma && !qos.IsValid()) {
+        LOG(ERROR) << "RDMA SL must be in [0,15] and traffic class in [0,255]";
+        return 1;
+    }
+
     // Initialize RDMA environment in advance.
     if (FLAGS_use_rdma) {
         brpc::rdma::GlobalRdmaInitializeOrDie();
diff --git a/example/rdma_performance/server.cpp b/example/rdma_performance/server.cpp
index 2e93e1eec7..b444b7b02d 100644
--- a/example/rdma_performance/server.cpp
+++ b/example/rdma_performance/server.cpp
@@ -28,6 +28,8 @@
 
 DEFINE_int32(port, 8002, "TCP Port of this server");
 DEFINE_bool(use_rdma, true, "Use RDMA or not");
+DEFINE_int32(rdma_sl, 0, "Outbound RDMA service level [0,15]");
+DEFINE_int32(rdma_traffic_class, 0, "Outbound RDMA traffic class byte [0,255]");
 
 butil::atomic<uint64_t> g_last_time(0);
 
@@ -77,6 +79,8 @@ int main(int argc, char* argv[]) {
 
     brpc::ServerOptions options;
     options.socket_mode = FLAGS_use_rdma? brpc::SOCKET_MODE_RDMA : brpc::SOCKET_MODE_TCP;
+    options.rdma_qos.service_level = FLAGS_rdma_sl;
+    options.rdma_qos.traffic_class = FLAGS_rdma_traffic_class;
     if (server.Start(FLAGS_port, &options) != 0) {
         LOG(ERROR) << "Fail to start EchoServer";
         return -1;
diff --git a/example/urma_performance/client.cpp b/example/urma_performance/client.cpp
index 9449dbe960..37aa844bec 100644
--- a/example/urma_performance/client.cpp
+++ b/example/urma_performance/client.cpp
@@ -44,6 +44,8 @@ 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 +90,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 +114,10 @@ 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,6 +138,8 @@ int main(int argc, char* argv[]) {
     }
     LOG(INFO) << "URMA performance client started (server=" << FLAGS_server
               << ", use_urma=" << FLAGS_use_urma
+              << ", urma_priority=" << FLAGS_urma_priority
+              << ", urma_sl=" << FLAGS_urma_sl
               << ", threads=" << thread_num
               << ", rpc_timeout_ms=" << FLAGS_rpc_timeout_ms << ")";
     while (!brpc::IsAskedToQuit()) {
diff --git a/example/urma_performance/server.cpp b/example/urma_performance/server.cpp
index baaff323b6..58c0d2ccaf 100644
--- a/example/urma_performance/server.cpp
+++ b/example/urma_performance/server.cpp
@@ -30,6 +30,8 @@
 
 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 +75,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 +84,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..9c5504f207 100644
--- a/src/brpc/acceptor.cpp
+++ b/src/brpc/acceptor.cpp
@@ -338,6 +338,8 @@ 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.rdma_qos = am->_rdma_qos;
+        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..b474fdd5a0 100644
--- a/src/brpc/acceptor.h
+++ b/src/brpc/acceptor.h
@@ -24,6 +24,8 @@
 #include "butil/containers/flat_map.h"
 #include "brpc/input_messenger.h"
 #include "brpc/socket_mode.h"
+#include "brpc/rdma/rdma_qos.h"
+#include "brpc/urma/urma_qos.h"
 
 
 namespace brpc {
@@ -137,6 +139,8 @@ friend class Server;
 
     // Choose to use a certain socket: 0 TCP, 1 RDMA
     SocketMode _socket_mode;
+    RdmaQosOptions _rdma_qos;
+    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 de0a1bde49..3c2ee44680 100644
--- a/src/brpc/channel.cpp
+++ b/src/brpc/channel.cpp
@@ -78,7 +78,8 @@ ChannelSSLOptions* ChannelOptions::mutable_ssl_options() {
 }
 
 static ChannelSignature ComputeChannelSignature(const ChannelOptions& opt) {
-    if (opt.auth == nullptr &&
+    if (opt.socket_mode == SOCKET_MODE_TCP &&
+        opt.auth == nullptr &&
         !opt.has_ssl_options() &&
         opt.client_host.empty() &&
         opt.device_name.empty() &&
@@ -139,8 +140,12 @@ static ChannelSignature ComputeChannelSignature(const ChannelOptions& opt) {
         }
         if (opt.socket_mode == SOCKET_MODE_RDMA) {
             buf.append("|rdma");
+            buf.append("|sl=").append(std::to_string(opt.rdma_qos.service_level));
+            buf.append("|tc=").append(std::to_string(opt.rdma_qos.traffic_class));
         } 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 +192,15 @@ int Channel::InitChannelOptions(const ChannelOptions* options) {
     if (options) {  // Override default options if user provided one.
         _options = *options;
     }
+    if (_options.socket_mode == SOCKET_MODE_RDMA && !_options.rdma_qos.IsValid()) {
+        LOG(ERROR) << "RDMA SL must be in [0,15] and traffic class in [0,255]";
+        return -1;
+    }
+    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 +421,8 @@ 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.rdma_qos = _options.rdma_qos;
+    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 +469,8 @@ 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.rdma_qos = _options.rdma_qos;
+    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..f11bdb04d7 100644
--- a/src/brpc/channel.h
+++ b/src/brpc/channel.h
@@ -38,6 +38,8 @@
 #include "brpc/naming_service_filter.h"
 #include "brpc/health_check_option.h"
 #include "brpc/socket_mode.h"
+#include "brpc/rdma/rdma_qos.h"
+#include "brpc/urma/urma_qos.h"
 
 namespace brpc {
 
@@ -111,6 +113,12 @@ struct ChannelOptions {
     // Default: SOCKET_MODE_TCP
     SocketMode socket_mode;
 
+    // Outbound RDMA QoS for this Channel. Used when socket_mode is RDMA.
+    RdmaQosOptions rdma_qos;
+
+    // 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/rdma/rdma_endpoint.cpp b/src/brpc/rdma/rdma_endpoint.cpp
index cb0fcc9d01..b9e47d1dcd 100644
--- a/src/brpc/rdma/rdma_endpoint.cpp
+++ b/src/brpc/rdma/rdma_endpoint.cpp
@@ -1533,9 +1533,9 @@ int RdmaEndpoint::BringUpQp(const ParsedHello& remote, bool is_server) {
     attr.ah_attr.grh.flow_label = 0;
     attr.ah_attr.grh.sgid_index = GetRdmaGidIndex();
     attr.ah_attr.grh.hop_limit = MAX_HOP_LIMIT;
-    attr.ah_attr.grh.traffic_class = 0;
+    attr.ah_attr.grh.traffic_class = _socket->rdma_qos().traffic_class;
     attr.ah_attr.dlid = remote.lid;
-    attr.ah_attr.sl = 0;
+    attr.ah_attr.sl = _socket->rdma_qos().service_level;
     attr.ah_attr.src_path_bits = 0;
     attr.ah_attr.static_rate = 0;
     attr.ah_attr.is_global = 1;
@@ -1952,6 +1952,8 @@ std::string RdmaEndpoint::GetStateStr() const {
 
 void RdmaEndpoint::DebugInfo(std::ostream& os, butil::StringPiece connector) const {
     os << "rdma_state=ON"
+       << connector << "rdma_sl=" << _socket->rdma_qos().service_level
+       << connector << "rdma_traffic_class=" << _socket->rdma_qos().traffic_class
        << connector << "handshake_state=" << GetStateStr()
        << connector << "handshake_version=" << static_cast<int>(_handshake_version)
        << connector << "rdma_sq_imm_window_size=" << _sq_imm_window_size
diff --git a/src/brpc/rdma/rdma_qos.h b/src/brpc/rdma/rdma_qos.h
new file mode 100644
index 0000000000..e75cdafa61
--- /dev/null
+++ b/src/brpc/rdma/rdma_qos.h
@@ -0,0 +1,36 @@
+// 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_RDMA_QOS_H
+#define BRPC_RDMA_QOS_H
+
+namespace brpc {
+
+// Outbound QP attributes. Their scheduling behavior is defined by the fabric.
+struct RdmaQosOptions {
+    int service_level = 0;  // SL, range [0, 15].
+    int traffic_class = 0;  // Full GRH/IP traffic class byte, range [0, 255].
+
+    bool IsValid() const {
+        return service_level >= 0 && service_level <= 15 &&
+               traffic_class >= 0 && traffic_class <= 255;
+    }
+};
+
+}  // namespace brpc
+
+#endif  // BRPC_RDMA_QOS_H
diff --git a/src/brpc/server.cpp b/src/brpc/server.cpp
index 19d58ac422..0564148d3f 100644
--- a/src/brpc/server.cpp
+++ b/src/brpc/server.cpp
@@ -875,6 +875,17 @@ 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_RDMA && !real_opt.rdma_qos.IsValid()) {
+        LOG(ERROR) << "RDMA SL must be in [0,15] and traffic class in [0,255]";
+        return -1;
+    }
+
+    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;
@@ -1162,6 +1173,8 @@ int Server::StartInternal(const butil::EndPoint& endpoint,
             _am->_socket_mode = _options.socket_mode;
             _am->_bthread_tag = _options.bthread_tag;
         }
+        _am->_rdma_qos = _options.rdma_qos;
+        _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..83e1d6027d 100644
--- a/src/brpc/server.h
+++ b/src/brpc/server.h
@@ -46,6 +46,8 @@
 #include "brpc/baidu_master_service.h"
 #include "brpc/rpc_pb_message_factory.h"
 #include "brpc/socket_mode.h"
+#include "brpc/rdma/rdma_qos.h"
+#include "brpc/urma/urma_qos.h"
 
 namespace brpc {
 
@@ -242,6 +244,12 @@ struct ServerOptions {
     // Default: SOCKET_MODE_TCP
     SocketMode socket_mode;
 
+    // Outbound RDMA QoS for responses on this server's accepted connections.
+    RdmaQosOptions rdma_qos;
+
+    // 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 57cfd2296a..4b49d9ea72 100644
--- a/src/brpc/socket.cpp
+++ b/src/brpc/socket.cpp
@@ -716,6 +716,14 @@ 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_RDMA && !options.rdma_qos.IsValid()) {
+        errno = EINVAL;
+        return -1;
+    }
+    if (options.socket_mode == SOCKET_MODE_URMA && !options.urma_qos.IsValid()) {
+        errno = EINVAL;
+        return -1;
+    }
     return VersionedRefWithId<Socket>::Create(id, options);
 }
 
@@ -731,6 +739,8 @@ int Socket::OnCreated(const SocketOptions& options) {
     });
     // start build the transport
     _socket_mode = options.socket_mode;
+    _rdma_qos = options.rdma_qos;
+    _urma_qos = options.urma_qos;
     _transport = TransportFactory::CreateTransport(options.socket_mode);
     CHECK(nullptr != _transport);
     _transport->Init(this, options);
@@ -2830,6 +2840,8 @@ int Socket::GetPooledSocket(SocketUniquePtr* pooled_socket) {
         opt.keytable_pool = _keytable_pool;
         opt.app_connect = _app_connect;
         opt.socket_mode = _socket_mode;
+        opt.rdma_qos = _rdma_qos;
+        opt.urma_qos = _urma_qos;
         socket_pool = new SocketPool(opt);
         SocketPool* expected = nullptr;
         if (!main_sp->socket_pool.compare_exchange_strong(
@@ -2936,6 +2948,8 @@ int Socket::GetShortSocket(SocketUniquePtr* short_socket) {
     opt.keytable_pool = _keytable_pool;
     opt.app_connect = _app_connect;
     opt.socket_mode = _socket_mode;
+    opt.rdma_qos = _rdma_qos;
+    opt.urma_qos = _urma_qos;
     if (get_client_side_messenger()->Create(opt, &id) != 0 ||
         Address(id, short_socket) != 0) {
         return -1;
diff --git a/src/brpc/socket.h b/src/brpc/socket.h
index da701e9ad7..8911b801a8 100644
--- a/src/brpc/socket.h
+++ b/src/brpc/socket.h
@@ -43,6 +43,8 @@
 #include "brpc/versioned_ref_with_id.h"
 #include "brpc/health_check_option.h"
 #include "brpc/socket_mode.h"
+#include "brpc/rdma/rdma_qos.h"
+#include "brpc/urma/urma_qos.h"
 #include "brpc/input_messenger_processor.h"  // InputMessengerProcessor
 
 namespace brpc {
@@ -309,6 +311,8 @@ struct SocketOptions {
     bool force_ssl{false};
     std::shared_ptr<SocketSSLContext> initial_ssl_ctx;
     SocketMode socket_mode{SOCKET_MODE_TCP};
+    RdmaQosOptions rdma_qos;
+    UrmaQosOptions urma_qos;
     bthread_keytable_pool_t* keytable_pool{nullptr};
     SocketConnection* conn{nullptr};
     std::shared_ptr<AppConnect> app_connect;
@@ -632,6 +636,8 @@ friend class TransportFactory;
 
     // The socket mode decided at creation.
     SocketMode socket_mode() const { return _socket_mode; }
+    const RdmaQosOptions& rdma_qos() const { return _rdma_qos; }
+    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.
@@ -999,6 +1005,8 @@ friend class TransportFactory;
 
     // Should use SOCKET_MODE_RDMA or SOCKET_MODE_TCP or Other, default is SOCKET_MODE_TCP Transport
     SocketMode _socket_mode;
+    RdmaQosOptions _rdma_qos;
+    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..4263b4d640 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;
 }
diff --git a/src/brpc/urma/urma_endpoint.cpp b/src/brpc/urma/urma_endpoint.cpp
index a2f163a349..3e01cc5b3a 100644
--- a/src/brpc/urma/urma_endpoint.cpp
+++ b/src/brpc/urma/urma_endpoint.cpp
@@ -393,6 +393,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;
@@ -405,7 +409,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) {
@@ -454,7 +459,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;
@@ -965,6 +970,16 @@ 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
@@ -1689,6 +1704,14 @@ void UrmaEndpoint::DebugInfo(std::ostream& os, butil::StringPiece) const {
        << " remote_recv_block_size=" << _remote_recv_block_size
        << " 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=" << rdma::RdmaFailureReasonName(_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 b9aa4b84c0..69abc78782 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,6 +713,7 @@ static bool GlobalUrmaInitializeImpl() {
               << " tp_type=" << (g_tp_type == URMA_CTP ? "ctp" : "rtp")
               << " priority=" << static_cast<unsigned>(g_jetty_priority)
               << " 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();
@@ -803,6 +805,103 @@ 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 b46475f3da..5dca8eeb49 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,20 @@ 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..2d28ff4434
--- /dev/null
+++ b/src/brpc/urma/urma_qos.h
@@ -0,0 +1,36 @@
+// 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 {
+    int priority = -1;  // -1 selects from SL or device default; explicit [0, 15].
+    int service_level = -1;  // -1 leaves SL unspecified; explicit range [0, 255].
+
+    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..a1e8f9b425 100644
--- a/src/brpc/urma_transport.cpp
+++ b/src/brpc/urma_transport.cpp
@@ -197,6 +197,11 @@ 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 +212,11 @@ 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 971fc91752..6801ec282e 100644
--- a/test/CMakeLists.txt
+++ b/test/CMakeLists.txt
@@ -295,6 +295,18 @@ foreach(BRPC_UT ${BRPC_UNITTESTS})
         brpc_test_config
         ${GPERFTOOLS_LIBRARIES})
     add_test(NAME ${BRPC_UT_WE} COMMAND ${BRPC_UT_WE})
+    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_rdma_qos_unittest.cpp b/test/brpc_rdma_qos_unittest.cpp
new file mode 100644
index 0000000000..06b9778d8c
--- /dev/null
+++ b/test/brpc_rdma_qos_unittest.cpp
@@ -0,0 +1,286 @@
+// 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 <netinet/in.h>
+#include <sys/socket.h>
+#include <gtest/gtest.h>
+#include <gflags/gflags.h>
+#if BRPC_WITH_RDMA
+#include <errno.h>
+#include <unistd.h>
+#include <functional>
+#include <utility>
+#include <vector>
+#include "butil/endpoint.h"
+#include "butil/fd_guard.h"
+#include "butil/time.h"
+#include "brpc/acceptor.h"
+#include "brpc/channel.h"
+#include "brpc/input_messenger.h"
+#include "brpc/server.h"
+#include "brpc/socket.h"
+#include "brpc/rdma_transport.h"
+#include "brpc/rdma/rdma_endpoint.h"
+#include "brpc/rdma/rdma_handshake.h"
+#include "brpc/rdma/rdma_helper.h"
+#include "brpc/details/load_balancer_with_naming.h"
+#include "butil/files/temp_file.h"
+
+using namespace brpc;
+
+namespace brpc {
+namespace rdma {
+extern butil::atomic<bool> g_rdma_available;
+extern bool g_skip_rdma_init;
+extern int (*IbvModifyQp)(ibv_qp*, ibv_qp_attr*, ibv_qp_attr_mask);
+}  // namespace rdma
+}  // namespace brpc
+
+static butil::EndPoint g_ep;
+static const int64_t WAIT_TIMEOUT_US = 5000000;
+static const int64_t WAIT_INTERVAL_US = 1000;
+
+static bool WaitUntil(const std::function<bool()>& pred,
+                      int64_t timeout_us = WAIT_TIMEOUT_US) {
+    const int64_t deadline = butil::gettimeofday_us() + timeout_us;
+    while (!pred()) {
+        if (butil::gettimeofday_us() >= deadline) {
+            return false;
+        }
+        usleep((useconds_t)WAIT_INTERVAL_US);
+    }
+    return true;
+}
+
+static void ConnectToServer(butil::fd_guard* sockfd) {
+    sockaddr_in addr;
+    bzero((char*)&addr, sizeof(addr));
+    addr.sin_family = AF_INET;
+    addr.sin_port = htons(g_ep.port);
+    addr.sin_addr = g_ep.ip;
+    sockfd->reset(socket(AF_INET, SOCK_STREAM, 0));
+    ASSERT_TRUE(*sockfd >= 0);
+    timeval tv;
+    tv.tv_sec = WAIT_TIMEOUT_US / 1000000;
+    tv.tv_usec = WAIT_TIMEOUT_US % 1000000;
+    ASSERT_EQ(0, setsockopt(*sockfd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)));
+    ASSERT_EQ(0, connect(*sockfd, (sockaddr*)&addr, sizeof(addr)));
+}
+
+class RdmaQosTest : public ::testing::Test {
+protected:
+    RdmaQosTest() {
+        EXPECT_EQ(0, butil::str2endpoint("127.0.0.1:8123", &g_ep));
+        EXPECT_EQ(0, _server_list.save(butil::endpoint2str(g_ep).c_str()));
+        _naming_url = std::string("File://") + _server_list.fname();
+    }
+
+    void TearDown() override { StopServer(); }
+
+    void StopServer() {
+        _server.Stop(0);
+        _server.Join();
+    }
+
+    Socket* GetSocketFromServer(size_t index) {
+        std::vector<SocketId> sids;
+        _server._am->ListConnections(&sids);
+        if (index >= sids.size()) {
+            return nullptr;
+        }
+        SocketUniquePtr socket;
+        return Socket::Address(sids[index], &socket) == 0 ? socket.get() : nullptr;
+    }
+
+    Socket* WaitForServerSocket() {
+        Socket* socket = nullptr;
+        WaitUntil([this, &socket] {
+            return (socket = GetSocketFromServer(0)) != nullptr;
+        });
+        return socket;
+    }
+
+    bool WaitForServerSocketGone() {
+        return WaitUntil([this] { return GetSocketFromServer(0) == nullptr; });
+    }
+
+    butil::TempFile _server_list;
+    std::string _naming_url;
+    Server _server;
+};
+
+static RdmaTransport* RdmaTransportOf(Socket* s) {
+    return static_cast<RdmaTransport*>(s->_transport.get());
+}
+static RdmaTransport* RdmaTransportOf(const SocketUniquePtr& s) {
+    return RdmaTransportOf(s.get());
+}
+
+TEST_F(RdmaQosTest, qos_rejects_invalid_options_before_connecting) {
+    for (const auto& values : {std::pair<int, int>(-1, 0), {16, 0}, {0, -1}, {0, 256}}) {
+        ChannelOptions options;
+        options.socket_mode = SOCKET_MODE_RDMA;
+        options.rdma_qos.service_level = values.first;
+        options.rdma_qos.traffic_class = values.second;
+        Channel channel;
+        EXPECT_EQ(-1, channel.Init(g_ep, &options));
+        ServerOptions server_options;
+        server_options.socket_mode = SOCKET_MODE_RDMA;
+        server_options.rdma_qos = options.rdma_qos;
+        Server server;
+        EXPECT_EQ(-1, server.Start("127.0.0.1:0", &server_options));
+        SocketOptions socket_options;
+        socket_options.socket_mode = SOCKET_MODE_RDMA;
+        socket_options.rdma_qos = options.rdma_qos;
+        SocketId sid;
+        EXPECT_EQ(-1, Socket::Create(socket_options, &sid));
+        EXPECT_EQ(EINVAL, errno);
+    }
+}
+
+TEST_F(RdmaQosTest, qos_isolates_channels_and_preserves_equal_config_sharing) {
+    ChannelOptions options;
+    options.socket_mode = SOCKET_MODE_RDMA;
+    Channel a, b, same, different_tc, default_rdma, tcp, tcp_with_qos;
+    ASSERT_EQ(0, default_rdma.Init(g_ep, &options));
+    ASSERT_EQ(0, tcp.Init(g_ep, nullptr));
+    EXPECT_NE(default_rdma._server_id, tcp._server_id);
+    options.rdma_qos.service_level = 3;
+    options.rdma_qos.traffic_class = 106;
+    ASSERT_EQ(0, a.Init(g_ep, &options));
+    ASSERT_EQ(0, same.Init(g_ep, &options));
+    EXPECT_EQ(a._server_id, same._server_id);
+    EXPECT_NE(a._server_id, default_rdma._server_id);
+    options.rdma_qos.service_level = 4;
+    ASSERT_EQ(0, b.Init(g_ep, &options));
+    EXPECT_NE(a._server_id, b._server_id);
+    options.rdma_qos.service_level = 3;
+    options.rdma_qos.traffic_class = 104;
+    ASSERT_EQ(0, different_tc.Init(g_ep, &options));
+    EXPECT_NE(a._server_id, different_tc._server_id);
+    options.socket_mode = SOCKET_MODE_TCP;
+    ASSERT_EQ(0, tcp_with_qos.Init(g_ep, &options));
+    EXPECT_EQ(tcp._server_id, tcp_with_qos._server_id);
+    SocketUniquePtr socket;
+    ASSERT_EQ(0, Socket::Address(a._server_id, &socket));
+    EXPECT_EQ(3, socket->rdma_qos().service_level);
+    EXPECT_EQ(106, socket->rdma_qos().traffic_class);
+}
+
+TEST_F(RdmaQosTest, qos_follows_pooled_short_and_named_connections) {
+    ChannelOptions options;
+    options.socket_mode = SOCKET_MODE_RDMA;
+    options.rdma_qos.service_level = 7;
+    options.rdma_qos.traffic_class = 255;
+    Channel direct, named;
+    ASSERT_EQ(0, direct.Init(g_ep, &options));
+    ASSERT_EQ(0, named.Init(_naming_url.c_str(), "rr", &options));
+    SocketUniquePtr main_socket;
+    ASSERT_EQ(0, Socket::Address(direct._server_id, &main_socket));
+    SocketUniquePtr pooled, short_socket, selected;
+    ASSERT_EQ(0, main_socket->GetPooledSocket(&pooled));
+    ASSERT_EQ(0, main_socket->GetShortSocket(&short_socket));
+    LoadBalancer::SelectIn input{};
+    LoadBalancer::SelectOut output(&selected);
+    ASSERT_EQ(0, named._lb->SelectServer(input, &output));
+    for (Socket* s : {pooled.get(), short_socket.get(), selected.get()}) {
+        EXPECT_EQ(7, s->rdma_qos().service_level);
+        EXPECT_EQ(255, s->rdma_qos().traffic_class);
+    }
+    EXPECT_EQ(main_socket->id(), selected->id());
+    EXPECT_EQ(0, pooled->ReturnToPool());
+    short_socket->SetFailed();
+}
+
+TEST_F(RdmaQosTest, qos_server_options_follow_accept_and_restart) {
+    for (int sl : {3, 7}) {
+        ServerOptions options;
+        options.socket_mode = SOCKET_MODE_RDMA;
+        options.rdma_qos.service_level = sl;
+        options.rdma_qos.traffic_class = 106;
+        ASSERT_EQ(0, _server.Start("127.0.0.1:0", &options));
+        g_ep = _server.listen_address();
+        butil::fd_guard fd;
+        ASSERT_NO_FATAL_FAILURE(ConnectToServer(&fd));
+        Socket* socket = WaitForServerSocket();
+        ASSERT_TRUE(socket != nullptr);
+        EXPECT_EQ(sl, socket->rdma_qos().service_level);
+        EXPECT_EQ(106, socket->rdma_qos().traffic_class);
+        fd.reset(-1);
+        ASSERT_TRUE(WaitForServerSocketGone());
+        StopServer();
+    }
+}
+
+static ibv_ah_attr g_qos_rtr_address;
+static int g_qos_rtr_calls;
+static int CaptureQosModify(ibv_qp*, ibv_qp_attr* attr, ibv_qp_attr_mask mask) {
+    if (mask & IBV_QP_AV) {
+        g_qos_rtr_address = attr->ah_attr;
+        ++g_qos_rtr_calls;
+    }
+    return 0;
+}
+
+TEST_F(RdmaQosTest, qos_applies_to_rtr_on_each_qp_bringup) {
+    for (const auto& values : {std::pair<int, int>(15, 255), {0, 0}}) {
+        SocketOptions options;
+        options.socket_mode = SOCKET_MODE_RDMA;
+        options.rdma_qos.service_level = values.first;
+        options.rdma_qos.traffic_class = values.second;
+        SocketId sid;
+        ASSERT_EQ(0, Socket::Create(options, &sid));
+        SocketUniquePtr socket;
+        ASSERT_EQ(0, Socket::Address(sid, &socket));
+        auto* ep = RdmaTransportOf(socket)->_rdma_ep;
+        ibv_qp qp{};
+        rdma::RdmaResource resource;
+        resource.qp = &qp;
+        ep->_resource = &resource;
+        ep->_rq_size = 0;
+        rdma::ParsedHello remote{};
+        const bool skip = rdma::g_skip_rdma_init;
+        const auto modify = rdma::IbvModifyQp;
+        rdma::g_skip_rdma_init = false;
+        rdma::IbvModifyQp = CaptureQosModify;
+        g_qos_rtr_calls = 0;
+        // Reapply attributes on both client and server bringup.
+        EXPECT_EQ(0, ep->BringUpQp(remote, false));
+        EXPECT_EQ(0, ep->BringUpQp(remote, true));
+        rdma::g_skip_rdma_init = skip;
+        rdma::IbvModifyQp = modify;
+        EXPECT_EQ(2, g_qos_rtr_calls);
+        EXPECT_EQ(values.first, g_qos_rtr_address.sl);
+        EXPECT_EQ(values.second, g_qos_rtr_address.grh.traffic_class);
+        ep->_resource = nullptr;
+        resource.qp = nullptr;
+        socket->SetFailed();
+    }
+}
+
+#endif  // BRPC_WITH_RDMA
+
+int main(int argc, char** argv) {
+    testing::InitGoogleTest(&argc, argv);
+    GFLAGS_NAMESPACE::ParseCommandLineFlags(&argc, &argv, true);
+#if BRPC_WITH_RDMA
+    rdma::g_rdma_available.store(true, butil::memory_order_relaxed);
+    rdma::g_skip_rdma_init = true;
+#endif
+    return RUN_ALL_TESTS();
+}
diff --git a/test/brpc_urma_qos_churn_unittest.cpp b/test/brpc_urma_qos_churn_unittest.cpp
new file mode 100644
index 0000000000..f06096a8fa
--- /dev/null
+++ b/test/brpc_urma_qos_churn_unittest.cpp
@@ -0,0 +1,264 @@
+// 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..ad78f30767
--- /dev/null
+++ b/test/brpc_urma_qos_unittest.cpp
@@ -0,0 +1,697 @@
+// 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 "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, 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();
+}