You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by "imjyz (via GitHub)" <gi...@apache.org> on 2026/04/23 03:33:59 UTC
[I] [Feature] Support extensible external configuration for RocketMQ clients, similar to Pulsar's `producer.properties` and `consumer.properties` [rocketmq-apis]
imjyz opened a new issue, #107:
URL: https://github.com/apache/rocketmq-apis/issues/107
### English
Apache Pulsar supports external producer and consumer configuration through files such as `producer.properties` and `consumer.properties`.
RocketMQ clients currently do not provide a similar standardized and extensible configuration capability.
I would like to propose adding similar support to RocketMQ clients, while making the configuration model extensible across different scopes, for example:
- client scope
- producer scope
- consumer scope
This kind of configuration model has several practical benefits:
- deployment-time configuration can be changed without modifying application code
- producer and consumer settings can be managed in a more uniform way
- environment-specific settings are easier to maintain across development, testing, and production
- operations teams can manage client behavior more conveniently
- new client capabilities can be introduced through clear configuration scopes instead of ad hoc code changes
- cross-language SDKs can evolve toward a more consistent configuration model
---
### 中文
Apache Pulsar 支持通过 `producer.properties`、`consumer.properties` 这类文件对 Producer 和 Consumer 进行外部配置。
RocketMQ 客户端目前还不支持类似的标准化、可扩展配置能力。
我希望 RocketMQ 客户端也能支持类似方式，并让这套配置模型能够按不同维度进行扩展，例如：
- client 维度
- producer 维度
- consumer 维度
这种配置模型有几个比较实际的好处：
- 部署时可以调整配置，而不需要修改应用代码
- Producer 和 Consumer 配置可以用更统一的方式管理
- 开发、测试、生产等不同环境下的配置更容易维护
- 运维侧可以更方便地管理客户端行为
- 后续新增客户端能力时，可以按清晰的配置维度扩展，而不是零散地落到代码中
- 有助于多语言 SDK 逐步形成更一致的配置模型
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Feature] Support extensible external configuration for RocketMQ clients, similar to Pulsar's `producer.properties` and `consumer.properties` [rocketmq-apis]
Posted by "fanlunyong (via GitHub)" <gi...@apache.org>.
fanlunyong commented on issue #107:
URL: https://github.com/apache/rocketmq-apis/issues/107#issuecomment-4301561726
这是来自QQ邮箱的自动回复邮件。
您好，请您放心您的邮件我已经收到。我将在看过邮件后，尽快给您回复。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Feature] Support extensible external configuration for RocketMQ clients, similar to Pulsar's `producer.properties` and `consumer.properties` [rocketmq-apis]
Posted by "imjyz (via GitHub)" <gi...@apache.org>.
imjyz commented on issue #107:
URL: https://github.com/apache/rocketmq-apis/issues/107#issuecomment-4333249877
**设计方案如下**
---
<!-- obsidian --><h2 data-heading="1. 背景">1. 背景</h2>
<p>Pulsar 的 <code>producer.properties</code> / <code>consumer.properties</code> 提供了一类实用能力：producer / consumer 创建时携带自定义 key/value，broker 将其挂到在线 producer / consumer 运行态对象上，并通过 stats / admin 接口暴露。</p>
<p>这类属性不属于消息级属性，也不是持久化控制面配置。它们的主要价值是提升在线实例识别、排障归因、审计、灰度观测和后续管理扩展能力。</p>
<p>RocketMQ 5.x gRPC 协议已经有合适的承载点：</p>
<ul>
<li>client 启动后建立双向 <code>Telemetry</code> stream。</li>
<li>client 主动上报 <code>TelemetryCommand.settings</code>。</li>
<li><code>Settings</code> 已包含 <code>client_type</code>、<code>access_point</code>、publish / subscription 配置、<code>UA</code>、<code>Metric</code>。</li>
<li>client 身份已经通过 <code>x-mq-client-id</code> header 表达，协议状态码中也已有 <code>CLIENT_ID_REQUIRED</code>。</li>
</ul>
<p>相关 issue：</p>
<ul>
<li><a href="https://github.com/apache/rocketmq-apis/issues/105" class="external-link" target="_blank" rel="noopener nofollow" aria-label="https://github.com/apache/rocketmq-apis/issues/105" data-tooltip-position="top">apache/rocketmq-apis#105</a>：讨论在 <code>Settings</code> 增加 <code>client_id</code>。本方案不建议新增该字段，因为现有协议已经通过 <code>x-mq-client-id</code> header 传递 client id。</li>
<li><a href="https://github.com/apache/rocketmq-apis/issues/107" class="external-link" target="_blank" rel="noopener nofollow" aria-label="https://github.com/apache/rocketmq-apis/issues/107" data-tooltip-position="top">apache/rocketmq-apis#107</a>：希望 RocketMQ client 支持类似 Pulsar <code>producer.properties</code> / <code>consumer.properties</code> 的外部化、可扩展配置模型，并区分 client / producer / consumer scope。</li>
</ul>
<p>本文聚焦两件事：</p>
<ul>
<li>在协议和服务端运行态中支持 RocketMQ client instance 级属性。</li>
<li>明确“外部化配置装载”和“上报给 server 的客户端属性”是两层能力，避免把任意配置全量透传给 broker。</li>
</ul>
<h2 data-heading="2. 目标与非目标">2. 目标与非目标</h2>
<h3 data-heading="目标">目标</h3>
<ul>
<li>在线 client 可识别：能基于 <code>x-mq-client-id</code> 串联 client identity、client type、consumer group、订阅关系等已有信息，并通过 <code>client_properties</code> 补充 <code>app</code>、<code>env</code>、<code>version</code>、<code>instance</code>、<code>owner</code> 等业务侧 metadata。</li>
<li>排障可归因：topic / group 堆积、消费异常、请求失败或限流时，可以定位到应用、环境、版本和实例。</li>
<li>审计可扩展：server hook、interceptor、audit log 可以读取客户端属性。</li>
<li>管理可扩展：为未来定向配置下推、单 client instance 限流、连接治理和标准化查询能力提供基础字段。</li>
<li>向后兼容：老 client 不上报新字段时继续可用，client identity 继续使用现有 <code>x-mq-client-id</code> header。</li>
</ul>
<h3 data-heading="非目标">非目标</h3>
<ul>
<li>不作为消息级 properties。</li>
<li>不作为 broker 原生路由、负载均衡或 dispatch 条件。</li>
<li>不作为 ack、重试、不可见时间、消费位点等行为控制参数。</li>
<li>不作为持久化 group / subscription 配置。</li>
<li>不作为认证或授权依据。</li>
</ul>
<p>如果需要表达“某个 consumer group 长期属于哪个团队、用途或等级”，应单独设计 group / subscription 级持久 properties，而不是复用 client instance 级属性。</p>
<h2 data-heading="3. 语义边界">3. 语义边界</h2>
<p>本文中的 client instance 指 SDK 内部创建的一个独立客户端实例，通常对应一个 producer、push consumer 或 simple consumer 对象。每个 client instance 都有独立的 <code>clientId</code> 和 Telemetry stream；一个应用进程可以创建多个 client instance。</p>
<p>以 Java SDK 为例，<code>ProducerImpl</code>、<code>PushConsumerImpl</code>、<code>SimpleConsumerImpl</code> 都继承 <code>ClientImpl</code>，每个实例构造时都会生成一个新的 <code>ClientId</code>。因此：</p>
<ul>
<li>一个 app 进程内创建多个 producer / consumer 时，会对应多个 <code>clientId</code>。</li>
<li>proxy 侧 <code>GrpcClientSettingsManager</code> 按 <code>clientId -> Settings</code> 保存 settings。</li>
<li><code>client_properties</code> 实际是 RocketMQ client instance 级 metadata，当前会落到 producer / consumer 实例维度。</li>
<li>如果应用希望表达 app 进程级身份，需要在每个 client instance 上重复上报相同的 <code>app</code>、<code>env</code>、<code>version</code>、<code>instance</code> 等字段，并在查询侧按这些字段聚合。</li>
</ul>
<p>该语义与 Pulsar <code>producer.properties</code> / <code>consumer.properties</code> 的连接或实例级语义接近，不应理解为“一个应用进程只有一份 properties”。</p>
<h2 data-heading="4. 命名与字段位置">4. 命名与字段位置</h2>
<p>#105 提出在 <code>UserAgent</code> 中增加 <code>extra_info</code>，用于承载 hostname、deployment environment、business tags 等客户端元数据。这个方向能解决“额外 metadata 无需频繁改 proto”的问题，但不适合作为本方案的首选落点。</p>
<p>本方案建议使用 <code>Settings.client_properties</code>，原因如下。</p>
<p>首先，<code>UserAgent</code> 当前语义是 SDK / runtime 信息，例如 language、version、platform，描述的是“这个客户端 SDK 是什么”；而本方案要表达的是“这个 client instance 属于哪个应用、环境、版本或部署单元”。将 <code>app</code>、<code>env</code>、<code>owner</code>、<code>deployment</code> 等业务侧属性放入 <code>UA.extra_info</code>，会混淆 user-agent 信息和 client metadata。</p>
<p>其次，<code>TelemetryCommand.settings</code> 是 client 建立会话后上报配置和运行态信息的入口。<code>client_properties</code> 与 client type、access point、publishing / subscription settings 同处 <code>Settings</code>，更符合“会话级 client metadata”的语义。<code>properties</code> 也与 Pulsar <code>producer.properties</code> / <code>consumer.properties</code> 的用户心智一致。</p>
<p>最后，独立的 <code>client_properties</code> 字段更容易在 proto 注释、client 校验和 server 校验中约束大小、key/value 格式、系统预留命名空间、敏感信息和禁止用途。<code>client_id</code> 仍继续通过 <code>x-mq-client-id</code> header 传递，不通过 <code>client_properties</code> 承载，也不在本方案中新增 <code>Settings.client_id</code>。</p>
<p>因此推荐字段为：</p>
<pre><code class="language-protobuf">map<string, string> client_properties = 9;
</code></pre>
<p>中文概念使用“客户端属性”，更准确地说是“RocketMQ client instance 级属性”。</p>
<h2 data-heading="5. 协议设计">5. 协议设计</h2>
<p>推荐在 <code>apache/rocketmq/v2/definition.proto</code> 的 <code>Settings</code> 中新增字段：</p>
<pre><code class="language-protobuf">message Settings {
optional ClientType client_type = 1;
optional Endpoints access_point = 2;
optional RetryPolicy backoff_policy = 3;
optional google.protobuf.Duration request_timeout = 4;
oneof pub_sub {
Publishing publishing = 5;
Subscription subscription = 6;
}
// User agent details.
UA user_agent = 7;
Metric metric = 8;
// Application-defined metadata associated with this client instance.
// These properties are for observability, troubleshooting, auditing and
// management only.
//
// Recommended keys: app, env, region, zone, version, instance, owner, deployment.
// Reserved prefix: "rocketmq." for system use.
//
// Size limits (server SHOULD enforce):
// - Maximum entries: 32
// - Maximum key length: 64 characters
// - Maximum value length: 256 characters
// - Total size: 4 KiB
map<string, string> client_properties = 9;
}
</code></pre>
<p>字段编号选择：</p>
<ul>
<li>使用 field number <code>9</code>，符合 RocketMQ StyleGuide 要求。</li>
<li>当前 <code>Settings</code> 已使用 <code>1-8</code>，<code>9</code> 是下一个可用编号。</li>
</ul>
<p>兼容性：</p>
<ul>
<li>proto3 新增 map 字段向后兼容。</li>
<li>老 client 不发送新字段时，server 读取到空 map。</li>
<li>老 server 不识别新字段时，按 unknown fields 忽略，不影响现有功能。</li>
<li>client identity 继续沿用 <code>x-mq-client-id</code> header，保持现有机制不变。</li>
</ul>
<h2 data-heading="6. Java Client 设计">6. Java Client 设计</h2>
<p>当前 Java SDK 关键链路：</p>
<ul>
<li>公共配置入口是 <code>ClientConfiguration</code> / <code>ClientConfigurationBuilder</code>。</li>
<li>每个 <code>ClientImpl</code> 实例构造时生成一个新的 <code>ClientId</code>。</li>
<li><code>Signature.sign()</code> 继续写入 <code>x-mq-client-id</code> header。</li>
<li><code>ClientSessionImpl.syncSettings0()</code> 发送 <code>TelemetryCommand.settings</code>。</li>
<li><code>PublishingSettings</code>、<code>PushSubscriptionSettings</code>、<code>SimpleSubscriptionSettings</code> 负责生成 proto <code>Settings</code>。</li>
</ul>
<p>推荐 API：</p>
<pre><code class="language-java">ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
.setEndpoints("rmq.example.com:8081")
.setClientProperty("app", "order-service")
.setClientProperty("env", "prod")
.setClientProperty("region", "cn-hangzhou")
.setClientProperty("version", "2.3.7")
.build();
</code></pre>
<p>改造步骤：</p>
<ul>
<li><code>ClientConfiguration</code> 增加 immutable <code>Map<String, String> clientProperties</code> 字段，构建 producer / consumer 时随对应 client instance 上报。</li>
<li><code>ClientConfigurationBuilder</code> 增加以下方法：
<ul>
<li><code>setClientProperty(String key, String value)</code>：设置单个属性。</li>
<li><code>setClientProperties(Map<String, String> properties)</code>：批量设置属性，覆盖已有属性。</li>
<li><code>removeClientProperty(String key)</code>：删除单个属性。</li>
<li><code>clearClientProperties()</code>：清空所有属性。</li>
</ul>
</li>
<li>抽象类 <code>org.apache.rocketmq.client.java.impl.Settings</code> 增加 <code>Map<String, String> clientProperties</code> 字段。</li>
<li><code>PublishingSettings.toProtobuf()</code>、<code>PushSubscriptionSettings.toProtobuf()</code>、<code>SimpleSubscriptionSettings.toProtobuf()</code> 写入 <code>.putAllClientProperties(clientProperties)</code>。</li>
<li><code>Signature.sign()</code> 保留 <code>x-mq-client-id</code>，用于兼容老 server。</li>
<li>client 侧做基础校验，避免明显无效请求进入网络链路。</li>
</ul>
<p>客户端约束：</p>
<ul>
<li>key 不能为空，只允许 <code>[a-zA-Z0-9_.-]</code>，长度不超过 64 字符。</li>
<li><code>rocketmq.</code> 前缀作为系统预留命名空间，业务侧不应使用；SDK 不应把它作为普通业务属性生成。</li>
<li>value 不能为 <code>null</code>，长度不超过 256 字符；空字符串允许通过。</li>
<li>properties 最多 32 项，总大小不超过 4 KiB。</li>
<li>不符合大小、格式约束时，builder 抛出 <code>IllegalArgumentException</code>。</li>
<li><code>ClientConfiguration</code> 对外暴露不可变 map，避免构建后被调用方修改。</li>
</ul>
<h2 data-heading="7. 外部化配置与 #107 的关系">7. 外部化配置与 #107 的关系</h2>
<p>#107 提到的 <code>producer.properties</code> / <code>consumer.properties</code> 更接近 SDK 配置装载模型，不应等价为“<strong>把 properties 文件所有 key/value 原样上报给 broker</strong>”。</p>
<p>这里需要避免两个误解：</p>
<ul>
<li>不应把 properties 文件中的所有配置全量透传到 <code>Settings.client_properties</code>。</li>
<li>不应把 <code>endpoints</code>、<code>requestTimeout</code>、topic、consumer group、subscription 等强类型配置放进 <code>client_properties</code>。</li>
</ul>
<p>本方案只关注上报给 server 的 client instance 级属性。强类型配置仍应映射到 SDK 已有字段；只有 <code>app</code>、<code>env</code>、<code>region</code>、<code>version</code>、<code>instance</code>、<code>owner</code> 这类观测和审计字段才进入 <code>Settings.client_properties</code>。</p>
<h2 data-heading="8. Server 保存模型">8. Server 保存模型</h2>
<p>gRPC proxy 是第一落点：</p>
<pre><code class="language-text">GrpcMessagingApplication.telemetry()
-> ClientActivity.telemetry()
-> processAndWriteClientSettings()
-> processClientSettings()
-> GrpcClientSettingsManager.updateClientSettings()
</code></pre>
<p><code>GrpcClientSettingsManager</code> 当前已有在线 settings 存储，key 是 SDK 生成的 clientId：</p>
<pre><code class="language-java">protected static final Map<String, Settings> CLIENT_SETTINGS_MAP = new ConcurrentHashMap<>();
</code></pre>
<p>推荐处理：</p>
<ul>
<li><code>ClientActivity.processClientSettings()</code> 使用现有 header / context 中的 client id 作为索引。</li>
<li>校验 <code>client_properties</code> 大小、key/value 格式，并处理系统预留命名空间。</li>
<li><code>GrpcClientSettingsManager.updateClientSettings(ctx, clientId, settings)</code> 保存 raw settings。</li>
<li>stream 关闭、channel 关闭、client termination、idle cleanup 时删除 metadata。</li>
<li>这份 map 是 producer / consumer client instance 级视图，不是 app 级视图。</li>
<li>这里不直接写入 <code>ClientChannelInfo</code>。<code>GrpcClientSettingsManager</code> 应作为 gRPC client settings 的 source of truth；面向现有 admin 查询链路的 <code>ClientChannelInfo</code> 同步在查询接口章节单独处理。</li>
</ul>
<p>服务端校验建议：</p>
规则 | 约束
-- | --
最大 entry 数 | 32
key 长度 | 1-64 字符
key 字符集 | [a-zA-Z0-9_.-]
value 长度 | 0-256 字符
总大小 | 不超过 4 KiB
系统预留命名空间 | rocketmq.
<p>错误处理：</p>
<ul>
<li>当前协议没有 <code>INVALID_CLIENT_CONFIG</code>，可先复用 <code>BAD_REQUEST = 40000</code>。</li>
<li>如社区希望错误语义更精确，可新增 <code>INVALID_CLIENT_PROPERTIES = 40021</code>。</li>
<li>建议大小、格式等校验失败时拒绝本次 settings，并返回错误码；不要静默截断，因为截断会降低排障可信度。</li>
<li><code>rocketmq.</code> 前缀是系统预留命名空间，业务自定义属性不应使用；如果未来引入系统字段，应由 SDK 或 server 明确定义语义。</li>
</ul>
<h2 data-heading="9. Admin 查询接口">9. Admin 查询接口</h2>
<p>第一阶段预期复用现有 admin connection 查询链路，避免新增独立查询接口：</p>
<ul>
<li><code>RequestCode.GET_CONSUMER_CONNECTION_LIST</code></li>
<li><code>RequestCode.GET_PRODUCER_CONNECTION_LIST</code></li>
</ul>
<p>这条链路读取的是 <code>ProducerManager</code> / <code>ConsumerManager</code> 中的 <code>ClientChannelInfo</code>，不是 <code>GrpcClientSettingsManager.CLIENT_SETTINGS_MAP</code>，因此需要把 <code>client_properties</code> 作为查询用冗余字段同步到 <code>ClientChannelInfo</code>。</p>
<p>实现建议：</p>
<ul>
<li>producer 注册保存在 <code>ProducerManager.groupChannelTable</code>，类型为 <code>Map<String, Map<Channel, ClientChannelInfo>></code>。</li>
<li>consumer 注册保存在 <code>ConsumerGroupInfo.channelInfoTable</code>，类型为 <code>Map<Channel, ClientChannelInfo></code>。</li>
<li><code>ClientChannelInfo</code> 增加 <code>Map<String, String> clientProperties</code> 字段，并更新构造函数、getter、equals / hashCode 方法。</li>
<li>在 gRPC client 注册 producer / consumer 时，从当前 <code>TelemetryCommand.settings</code> 提取 <code>client_properties</code>，构造带 properties 的 <code>ClientChannelInfo</code>。</li>
<li><code>AdminBrokerProcessor.getConsumerConnectionList()</code> / <code>getProducerConnectionList()</code> 从 <code>ClientChannelInfo</code> 读取并填充 <code>Connection.clientProperties</code>。</li>
</ul>
<p>当前返回体中 <code>Connection</code> 只有 <code>clientId</code>、<code>clientAddr</code>、<code>language</code>、<code>version</code>。建议增加：</p>
<pre><code class="language-java">public class Connection {
private String clientId;
private String clientAddr;
private LanguageCode language;
private int version;
private Map<String, String> clientProperties;
}
</code></pre>
<p>现有 Java admin API 可复用：</p>
<ul>
<li><code>DefaultMQAdminExt.examineConsumerConnectionInfo(...)</code></li>
<li><code>DefaultMQAdminExt.examineProducerConnectionInfo(...)</code></li>
</ul>
<p>CLI 建议：</p>
<ul>
<li><code>mqadmin consumerConnection</code> / <code>producerConnection</code> 默认输出保持不变，避免破坏现有表格格式。</li>
<li>增加 <code>--verbose</code> 或 <code>--show-properties</code> 选项展示客户端属性。</li>
</ul>
<p>输出示例：</p>
<pre><code class="language-text">ClientId: xxx
ClientAddr: xxx
Language: JAVA
Version: 5.1.0
Properties:
app=order-service
env=prod
version=2.3.7
</code></pre>
<p>标准化 gRPC admin API 可作为第二阶段，在 <code>rocketmq-apis/admin.proto</code> 增加 query clients RPC。当前 <code>admin.proto</code> 较轻量，不建议第一阶段直接扩大协议面。</p>
<p>边界：</p>
<ul>
<li>第一阶段仅支持 gRPC client。</li>
<li>classic remoting client 不上报 <code>Settings.client_properties</code>，因此不会具备该能力。</li>
</ul>
<h2 data-heading="10. 落地顺序">10. 落地顺序</h2>
<h3 data-heading="阶段一：协议字段">阶段一：协议字段</h3>
<ul>
<li>在 <code>apache/rocketmq/v2/definition.proto</code> 的 <code>Settings</code> message 增加 <code>map<string, string> client_properties = 9;</code>。</li>
<li>添加 protobuf 注释，说明用途、限制、推荐字段和禁止用途。</li>
<li>不在 properties 特性中支持 <code>client_id</code>，client id 继续使用现有 <code>x-mq-client-id</code> header。</li>
<li>可选：如需要专用错误码，在 <code>Code</code> enum 增加 <code>INVALID_CLIENT_PROPERTIES = 40021</code>。</li>
</ul>
<h3 data-heading="阶段二：Java client">阶段二：Java client</h3>
<ul>
<li><code>ClientConfiguration</code> 增加 <code>Map<String, String> clientProperties</code> 字段，使用不可变 map。</li>
<li><code>ClientConfigurationBuilder</code> 增加 <code>setClientProperty</code>、<code>setClientProperties</code>、<code>removeClientProperty</code>、<code>clearClientProperties</code>。</li>
<li>builder 中实现 client 侧校验逻辑。</li>
<li><code>org.apache.rocketmq.client.java.impl.Settings</code> 抽象类增加 <code>Map<String, String> clientProperties</code> 字段。</li>
<li>三类 settings 的 <code>toProtobuf()</code> 写入 <code>client_properties</code>。</li>
<li>添加 builder 行为、immutability、校验逻辑和端到端上报测试。</li>
</ul>
<h3 data-heading="阶段三：Proxy runtime">阶段三：Proxy runtime</h3>
<ul>
<li><code>ClientActivity.processClientSettings()</code> 基于 header <code>x-mq-client-id</code> 索引并校验 properties。</li>
<li>实现 server 侧校验逻辑。</li>
<li>校验失败返回 <code>BAD_REQUEST</code> 或新增的 <code>INVALID_CLIENT_PROPERTIES</code> 错误码。</li>
<li><code>GrpcClientSettingsManager.updateClientSettings()</code> 保存带 properties 的 raw settings。</li>
<li>stream / channel 清理时移除 settings。</li>
<li>添加单元测试验证校验逻辑和边界条件。</li>
</ul>
<h3 data-heading="阶段四：Broker / admin 查询">阶段四：Broker / admin 查询</h3>
<ul>
<li>按第 9 节复用现有 connection 查询链路。</li>
<li>扩展 <code>ClientChannelInfo</code> 和 <code>Connection</code>，在 admin 查询结果中暴露 <code>clientProperties</code>。</li>
<li><code>mqadmin producerConnection</code> / <code>consumerConnection</code> 增加 <code>--verbose</code> 或 <code>--show-properties</code> 选项。</li>
<li>添加端到端查询测试。</li>
</ul>
<h3 data-heading="阶段五：标准化管理 API（可选）">阶段五：标准化管理 API（可选）</h3>
<ul>
<li>字段和使用方式稳定后，再考虑在 <code>rocketmq-apis/admin.proto</code> 增加 query clients API。</li>
<li>设计统一的 gRPC admin 接口用于查询 client metadata，并支持按 properties 过滤。</li>
</ul>
<h2 data-heading="11. 待讨论问题">11. 待讨论问题</h2>
<ol>
<li>是否需要新增专用错误码，还是复用 <code>BAD_REQUEST = 40000</code>？</li>
<li>是否需要支持 properties 动态更新，还是只在连接建立时设置一次？</li>
<li>长期是否需要新增 proxy 侧 client metadata 查询接口，替代 <code>ClientChannelInfo</code> 冗余同步？</li>
</ol>## 1. 背景
Pulsar 的 `producer.properties` / `consumer.properties` 提供了一类实用能力：producer / consumer 创建时携带自定义 key/value，broker 将其挂到在线 producer / consumer 运行态对象上，并通过 stats / admin 接口暴露。
这类属性不属于消息级属性，也不是持久化控制面配置。它们的主要价值是提升在线实例识别、排障归因、审计、灰度观测和后续管理扩展能力。
RocketMQ 5.x gRPC 协议已经有合适的承载点：
- client 启动后建立双向 `Telemetry` stream。
- client 主动上报 `TelemetryCommand.settings`。
- `Settings` 已包含 `client_type`、`access_point`、publish / subscription 配置、`UA`、`Metric`。
- client 身份已经通过 `x-mq-client-id` header 表达，协议状态码中也已有 `CLIENT_ID_REQUIRED`。
相关 issue：
- [[apache/rocketmq-apis#105](https://github.com/apache/rocketmq-apis/issues/105)](https://github.com/apache/rocketmq-apis/issues/105)：讨论在 `Settings` 增加 `client_id`。本方案不建议新增该字段，因为现有协议已经通过 `x-mq-client-id` header 传递 client id。
- [[apache/rocketmq-apis#107](https://github.com/apache/rocketmq-apis/issues/107)](https://github.com/apache/rocketmq-apis/issues/107)：希望 RocketMQ client 支持类似 Pulsar `producer.properties` / `consumer.properties` 的外部化、可扩展配置模型，并区分 client / producer / consumer scope。
本文聚焦两件事：
- 在协议和服务端运行态中支持 RocketMQ client instance 级属性。
- 明确“外部化配置装载”和“上报给 server 的客户端属性”是两层能力，避免把任意配置全量透传给 broker。
## 2. 目标与非目标
### 目标
- 在线 client 可识别：能基于 `x-mq-client-id` 串联 client identity、client type、consumer group、订阅关系等已有信息，并通过 `client_properties` 补充 `app`、`env`、`version`、`instance`、`owner` 等业务侧 metadata。
- 排障可归因：topic / group 堆积、消费异常、请求失败或限流时，可以定位到应用、环境、版本和实例。
- 审计可扩展：server hook、interceptor、audit log 可以读取客户端属性。
- 管理可扩展：为未来定向配置下推、单 client instance 限流、连接治理和标准化查询能力提供基础字段。
- 向后兼容：老 client 不上报新字段时继续可用，client identity 继续使用现有 `x-mq-client-id` header。
### 非目标
- 不作为消息级 properties。
- 不作为 broker 原生路由、负载均衡或 dispatch 条件。
- 不作为 ack、重试、不可见时间、消费位点等行为控制参数。
- 不作为持久化 group / subscription 配置。
- 不作为认证或授权依据。
如果需要表达“某个 consumer group 长期属于哪个团队、用途或等级”，应单独设计 group / subscription 级持久 properties，而不是复用 client instance 级属性。
## 3. 语义边界
本文中的 client instance 指 SDK 内部创建的一个独立客户端实例，通常对应一个 producer、push consumer 或 simple consumer 对象。每个 client instance 都有独立的 `clientId` 和 Telemetry stream；一个应用进程可以创建多个 client instance。
以 Java SDK 为例，`ProducerImpl`、`PushConsumerImpl`、`SimpleConsumerImpl` 都继承 `ClientImpl`，每个实例构造时都会生成一个新的 `ClientId`。因此：
- 一个 app 进程内创建多个 producer / consumer 时，会对应多个 `clientId`。
- proxy 侧 `GrpcClientSettingsManager` 按 `clientId -> Settings` 保存 settings。
- `client_properties` 实际是 RocketMQ client instance 级 metadata，当前会落到 producer / consumer 实例维度。
- 如果应用希望表达 app 进程级身份，需要在每个 client instance 上重复上报相同的 `app`、`env`、`version`、`instance` 等字段，并在查询侧按这些字段聚合。
该语义与 Pulsar `producer.properties` / `consumer.properties` 的连接或实例级语义接近，不应理解为“一个应用进程只有一份 properties”。
## 4. 命名与字段位置
#105 提出在 `UserAgent` 中增加 `extra_info`，用于承载 hostname、deployment environment、business tags 等客户端元数据。这个方向能解决“额外 metadata 无需频繁改 proto”的问题，但不适合作为本方案的首选落点。
本方案建议使用 `Settings.client_properties`，原因如下。
首先，`UserAgent` 当前语义是 SDK / runtime 信息，例如 language、version、platform，描述的是“这个客户端 SDK 是什么”；而本方案要表达的是“这个 client instance 属于哪个应用、环境、版本或部署单元”。将 `app`、`env`、`owner`、`deployment` 等业务侧属性放入 `UA.extra_info`，会混淆 user-agent 信息和 client metadata。
其次，`TelemetryCommand.settings` 是 client 建立会话后上报配置和运行态信息的入口。`client_properties` 与 client type、access point、publishing / subscription settings 同处 `Settings`，更符合“会话级 client metadata”的语义。`properties` 也与 Pulsar `producer.properties` / `consumer.properties` 的用户心智一致。
最后，独立的 `client_properties` 字段更容易在 proto 注释、client 校验和 server 校验中约束大小、key/value 格式、系统预留命名空间、敏感信息和禁止用途。`client_id` 仍继续通过 `x-mq-client-id` header 传递，不通过 `client_properties` 承载，也不在本方案中新增 `Settings.client_id`。
因此推荐字段为：
```protobuf
map<string, string> client_properties = 9;
```
中文概念使用“客户端属性”，更准确地说是“RocketMQ client instance 级属性”。
## 5. 协议设计
推荐在 `apache/rocketmq/v2/definition.proto` 的 `Settings` 中新增字段：
```protobuf
message Settings {
optional ClientType client_type = 1;
optional Endpoints access_point = 2;
optional RetryPolicy backoff_policy = 3;
optional google.protobuf.Duration request_timeout = 4;
oneof pub_sub {
Publishing publishing = 5;
Subscription subscription = 6;
}
// User agent details.
UA user_agent = 7;
Metric metric = 8;
// Application-defined metadata associated with this client instance.
// These properties are for observability, troubleshooting, auditing and
// management only.
//
// Recommended keys: app, env, region, zone, version, instance, owner, deployment.
// Reserved prefix: "rocketmq." for system use.
//
// Size limits (server SHOULD enforce):
// - Maximum entries: 32
// - Maximum key length: 64 characters
// - Maximum value length: 256 characters
// - Total size: 4 KiB
map<string, string> client_properties = 9;
}
```
字段编号选择：
- 使用 field number `9`，符合 RocketMQ StyleGuide 要求。
- 当前 `Settings` 已使用 `1-8`，`9` 是下一个可用编号。
兼容性：
- proto3 新增 map 字段向后兼容。
- 老 client 不发送新字段时，server 读取到空 map。
- 老 server 不识别新字段时，按 unknown fields 忽略，不影响现有功能。
- client identity 继续沿用 `x-mq-client-id` header，保持现有机制不变。
## 6. Java Client 设计
当前 Java SDK 关键链路：
- 公共配置入口是 `ClientConfiguration` / `ClientConfigurationBuilder`。
- 每个 `ClientImpl` 实例构造时生成一个新的 `ClientId`。
- `Signature.sign()` 继续写入 `x-mq-client-id` header。
- `ClientSessionImpl.syncSettings0()` 发送 `TelemetryCommand.settings`。
- `PublishingSettings`、`PushSubscriptionSettings`、`SimpleSubscriptionSettings` 负责生成 proto `Settings`。
推荐 API：
```java
ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
.setEndpoints("rmq.example.com:8081")
.setClientProperty("app", "order-service")
.setClientProperty("env", "prod")
.setClientProperty("region", "cn-hangzhou")
.setClientProperty("version", "2.3.7")
.build();
```
改造步骤：
- `ClientConfiguration` 增加 immutable `Map<String, String> clientProperties` 字段，构建 producer / consumer 时随对应 client instance 上报。
- `ClientConfigurationBuilder` 增加以下方法：
- `setClientProperty(String key, String value)`：设置单个属性。
- `setClientProperties(Map<String, String> properties)`：批量设置属性，覆盖已有属性。
- `removeClientProperty(String key)`：删除单个属性。
- `clearClientProperties()`：清空所有属性。
- 抽象类 `org.apache.rocketmq.client.java.impl.Settings` 增加 `Map<String, String> clientProperties` 字段。
- `PublishingSettings.toProtobuf()`、`PushSubscriptionSettings.toProtobuf()`、`SimpleSubscriptionSettings.toProtobuf()` 写入 `.putAllClientProperties(clientProperties)`。
- `Signature.sign()` 保留 `x-mq-client-id`，用于兼容老 server。
- client 侧做基础校验，避免明显无效请求进入网络链路。
客户端约束：
- key 不能为空，只允许 `[a-zA-Z0-9_.-]`，长度不超过 64 字符。
- `rocketmq.` 前缀作为系统预留命名空间，业务侧不应使用；SDK 不应把它作为普通业务属性生成。
- value 不能为 `null`，长度不超过 256 字符；空字符串允许通过。
- properties 最多 32 项，总大小不超过 4 KiB。
- 不符合大小、格式约束时，builder 抛出 `IllegalArgumentException`。
- `ClientConfiguration` 对外暴露不可变 map，避免构建后被调用方修改。
## 7. 外部化配置与 #107 的关系
#107 提到的 `producer.properties` / `consumer.properties` 更接近 SDK 配置装载模型，不应等价为“**把 properties 文件所有 key/value 原样上报给 broker**”。
这里需要避免两个误解：
- 不应把 properties 文件中的所有配置全量透传到 `Settings.client_properties`。
- 不应把 `endpoints`、`requestTimeout`、topic、consumer group、subscription 等强类型配置放进 `client_properties`。
本方案只关注上报给 server 的 client instance 级属性。强类型配置仍应映射到 SDK 已有字段；只有 `app`、`env`、`region`、`version`、`instance`、`owner` 这类观测和审计字段才进入 `Settings.client_properties`。
## 8. Server 保存模型
gRPC proxy 是第一落点：
```text
GrpcMessagingApplication.telemetry()
-> ClientActivity.telemetry()
-> processAndWriteClientSettings()
-> processClientSettings()
-> GrpcClientSettingsManager.updateClientSettings()
```
`GrpcClientSettingsManager` 当前已有在线 settings 存储，key 是 SDK 生成的 clientId：
```java
protected static final Map<String, Settings> CLIENT_SETTINGS_MAP = new ConcurrentHashMap<>();
```
推荐处理：
- `ClientActivity.processClientSettings()` 使用现有 header / context 中的 client id 作为索引。
- 校验 `client_properties` 大小、key/value 格式，并处理系统预留命名空间。
- `GrpcClientSettingsManager.updateClientSettings(ctx, clientId, settings)` 保存 raw settings。
- stream 关闭、channel 关闭、client termination、idle cleanup 时删除 metadata。
- 这份 map 是 producer / consumer client instance 级视图，不是 app 级视图。
- 这里不直接写入 `ClientChannelInfo`。`GrpcClientSettingsManager` 应作为 gRPC client settings 的 source of truth；面向现有 admin 查询链路的 `ClientChannelInfo` 同步在查询接口章节单独处理。
服务端校验建议：
| 规则 | 约束 |
| ---------- | ---------------- |
| 最大 entry 数 | 32 |
| key 长度 | 1-64 字符 |
| key 字符集 | `[a-zA-Z0-9_.-]` |
| value 长度 | 0-256 字符 |
| 总大小 | 不超过 4 KiB |
| 系统预留命名空间 | `rocketmq.` |
错误处理：
- 当前协议没有 `INVALID_CLIENT_CONFIG`，可先复用 `BAD_REQUEST = 40000`。
- 如社区希望错误语义更精确，可新增 `INVALID_CLIENT_PROPERTIES = 40021`。
- 建议大小、格式等校验失败时拒绝本次 settings，并返回错误码；不要静默截断，因为截断会降低排障可信度。
- `rocketmq.` 前缀是系统预留命名空间，业务自定义属性不应使用；如果未来引入系统字段，应由 SDK 或 server 明确定义语义。
## 9. Admin 查询接口
第一阶段预期复用现有 admin connection 查询链路，避免新增独立查询接口：
- `RequestCode.GET_CONSUMER_CONNECTION_LIST`
- `RequestCode.GET_PRODUCER_CONNECTION_LIST`
这条链路读取的是 `ProducerManager` / `ConsumerManager` 中的 `ClientChannelInfo`，不是 `GrpcClientSettingsManager.CLIENT_SETTINGS_MAP`，因此需要把 `client_properties` 作为查询用冗余字段同步到 `ClientChannelInfo`。
实现建议：
- producer 注册保存在 `ProducerManager.groupChannelTable`，类型为 `Map<String, Map<Channel, ClientChannelInfo>>`。
- consumer 注册保存在 `ConsumerGroupInfo.channelInfoTable`，类型为 `Map<Channel, ClientChannelInfo>`。
- `ClientChannelInfo` 增加 `Map<String, String> clientProperties` 字段，并更新构造函数、getter、equals / hashCode 方法。
- 在 gRPC client 注册 producer / consumer 时，从当前 `TelemetryCommand.settings` 提取 `client_properties`，构造带 properties 的 `ClientChannelInfo`。
- `AdminBrokerProcessor.getConsumerConnectionList()` / `getProducerConnectionList()` 从 `ClientChannelInfo` 读取并填充 `Connection.clientProperties`。
当前返回体中 `Connection` 只有 `clientId`、`clientAddr`、`language`、`version`。建议增加：
```java
public class Connection {
private String clientId;
private String clientAddr;
private LanguageCode language;
private int version;
private Map<String, String> clientProperties;
}
```
现有 Java admin API 可复用：
- `DefaultMQAdminExt.examineConsumerConnectionInfo(...)`
- `DefaultMQAdminExt.examineProducerConnectionInfo(...)`
CLI 建议：
- `mqadmin consumerConnection` / `producerConnection` 默认输出保持不变，避免破坏现有表格格式。
- 增加 `--verbose` 或 `--show-properties` 选项展示客户端属性。
输出示例：
```text
ClientId: xxx
ClientAddr: xxx
Language: JAVA
Version: 5.1.0
Properties:
app=order-service
env=prod
version=2.3.7
```
标准化 gRPC admin API 可作为第二阶段，在 `rocketmq-apis/admin.proto` 增加 query clients RPC。当前 `admin.proto` 较轻量，不建议第一阶段直接扩大协议面。
边界：
- 第一阶段仅支持 gRPC client。
- classic remoting client 不上报 `Settings.client_properties`，因此不会具备该能力。
## 10. 落地顺序
### 阶段一：协议字段
- 在 `apache/rocketmq/v2/definition.proto` 的 `Settings` message 增加 `map<string, string> client_properties = 9;`。
- 添加 protobuf 注释，说明用途、限制、推荐字段和禁止用途。
- 不在 properties 特性中支持 `client_id`，client id 继续使用现有 `x-mq-client-id` header。
- 可选：如需要专用错误码，在 `Code` enum 增加 `INVALID_CLIENT_PROPERTIES = 40021`。
### 阶段二：Java client
- `ClientConfiguration` 增加 `Map<String, String> clientProperties` 字段，使用不可变 map。
- `ClientConfigurationBuilder` 增加 `setClientProperty`、`setClientProperties`、`removeClientProperty`、`clearClientProperties`。
- builder 中实现 client 侧校验逻辑。
- `org.apache.rocketmq.client.java.impl.Settings` 抽象类增加 `Map<String, String> clientProperties` 字段。
- 三类 settings 的 `toProtobuf()` 写入 `client_properties`。
- 添加 builder 行为、immutability、校验逻辑和端到端上报测试。
### 阶段三：Proxy runtime
- `ClientActivity.processClientSettings()` 基于 header `x-mq-client-id` 索引并校验 properties。
- 实现 server 侧校验逻辑。
- 校验失败返回 `BAD_REQUEST` 或新增的 `INVALID_CLIENT_PROPERTIES` 错误码。
- `GrpcClientSettingsManager.updateClientSettings()` 保存带 properties 的 raw settings。
- stream / channel 清理时移除 settings。
- 添加单元测试验证校验逻辑和边界条件。
### 阶段四：Broker / admin 查询
- 按第 9 节复用现有 connection 查询链路。
- 扩展 `ClientChannelInfo` 和 `Connection`，在 admin 查询结果中暴露 `clientProperties`。
- `mqadmin producerConnection` / `consumerConnection` 增加 `--verbose` 或 `--show-properties` 选项。
- 添加端到端查询测试。
### 阶段五：标准化管理 API（可选）
- 字段和使用方式稳定后，再考虑在 `rocketmq-apis/admin.proto` 增加 query clients API。
- 设计统一的 gRPC admin 接口用于查询 client metadata，并支持按 properties 过滤。
## 11. 待讨论问题
1. 是否需要新增专用错误码，还是复用 `BAD_REQUEST = 40000`？
2. 是否需要支持 properties 动态更新，还是只在连接建立时设置一次？
3. 长期是否需要新增 proxy 侧 client metadata 查询接口，替代 `ClientChannelInfo` 冗余同步？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Feature] Support extensible external configuration for RocketMQ clients, similar to Pulsar's `producer.properties` and `consumer.properties` [rocketmq-apis]
Posted by "RockteMQ-AI (via GitHub)" <gi...@apache.org>.
RockteMQ-AI commented on issue #107:
URL: https://github.com/apache/rocketmq-apis/issues/107#issuecomment-5343265432
**Issue Evaluation**
Category: `enhancement` | Status: **Evaluated**
This proposes adding extensible external configuration support for RocketMQ clients, similar to Pulsar's `producer.properties` and `consumer.properties`.
**Feasibility:** Feasible but requires cross-language coordination (Java/Go/C++/C#/Rust SDKs).
**Scope:** Client SDK configuration layer across all language implementations; may need proto definition changes in `Settings` or `UserAgent`.
**Compatibility:** Must be backward-compatible — external config should be optional and not break existing programmatic configuration.
This is a valuable enhancement for production deployments where configuration is managed externally (ConfigMap, environment files, etc.). The extensibility model should be carefully designed to avoid fragmentation across language SDKs.
---
*Automated evaluation by RockteMQ-AI*
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org