You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by "wang-jiahua (via GitHub)" <gi...@apache.org> on 2026/06/22 10:01:00 UTC
[PR] [ISSUE #10538] Eliminate SendMessageRequestHeaderV2 intermediate object in client [rocketmq]
wang-jiahua opened a new pull request, #10540:
URL: https://github.com/apache/rocketmq/pull/10540
## Which Issue(s) This PR Fixes
Fixes #10538
## Brief Description
- MQClientAPIImpl: skip createSendMessageRequestHeaderV2() for SEND_MESSAGE_V2
- MQClientAPIExt: skip createSendMessageRequestHeaderV2() for SEND_MESSAGE_V2_WITH_REPLY
- Eliminates per-send V2 object allocation + 14 field copies
- Depends on #10535 (R2 Send header FastCodesHeader) for V2 short-key encode
## How Did You Test This Change
CI checkstyle + compilation passed. Server benchmark pending (depends on #10535 merge).
--
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: [PR] [ISSUE #10538] Eliminate SendMessageRequestHeaderV2 intermediate object in client [rocketmq]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #10540:
URL: https://github.com/apache/rocketmq/pull/10540#discussion_r3503063597
##########
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/SendMessageRequestHeader.java:
##########
@@ -183,22 +190,140 @@ public void setBatch(Boolean batch) {
}
public static SendMessageRequestHeader parseRequestHeader(RemotingCommand request) throws RemotingCommandException {
- SendMessageRequestHeaderV2 requestHeaderV2 = null;
- SendMessageRequestHeader requestHeader = null;
- switch (request.getCode()) {
- case RequestCode.SEND_BATCH_MESSAGE:
- case RequestCode.SEND_MESSAGE_V2:
- requestHeaderV2 = request.decodeCommandCustomHeader(SendMessageRequestHeaderV2.class);
- case RequestCode.SEND_MESSAGE:
- if (null == requestHeaderV2) {
- requestHeader = request.decodeCommandCustomHeader(SendMessageRequestHeader.class);
- } else {
- requestHeader = SendMessageRequestHeaderV2.createSendMessageRequestHeaderV1(requestHeaderV2);
- }
- default:
- break;
- }
- return requestHeader;
+ return request.decodeCommandCustomHeader(SendMessageRequestHeader.class);
+ }
+
+ @Override
+ public void encode(ByteBuf out) {
+ writeIfNotNull(out, "a", producerGroup);
+ writeIfNotNull(out, "b", topic);
+ writeIfNotNull(out, "c", defaultTopic);
+ writeIfNotNull(out, "d", defaultTopicQueueNums);
+ writeIfNotNull(out, "e", queueId);
+ writeIfNotNull(out, "f", sysFlag);
+ writeIfNotNull(out, "g", bornTimestamp);
+ writeIfNotNull(out, "h", flag);
+ writeIfNotNull(out, "i", properties);
+ writeIfNotNull(out, "j", reconsumeTimes);
+ writeIfNotNull(out, "k", unitMode);
+ writeIfNotNull(out, "l", maxReconsumeTimes);
+ writeIfNotNull(out, "m", batch);
+ writeIfNotNull(out, "n", getBname());
+ }
Review Comment:
SendMessageRequestHeader.encode() now serializes only short V2 keys ("a".."n"). For RequestCode.SEND_MESSAGE, the broker-side auth path (DefaultAuthorizationContextBuilder) reads request.getExtFields() directly and expects the long key "topic" (and old brokers using reflection-based header decode expect long keys like "producerGroup", "sysFlag", etc.). With fast ROCKETMQ encoding, extFields won’t contain those long keys, so SEND_MESSAGE authorization/topic parsing can break and older brokers may fail to decode required fields.
To preserve V1 compatibility, encode both the short V2 keys and the original long field names (at least for all `@CFNotNull` fields, plus topic).
##########
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/SendMessageRequestHeader.java:
##########
@@ -183,22 +190,140 @@ public void setBatch(Boolean batch) {
}
public static SendMessageRequestHeader parseRequestHeader(RemotingCommand request) throws RemotingCommandException {
- SendMessageRequestHeaderV2 requestHeaderV2 = null;
- SendMessageRequestHeader requestHeader = null;
- switch (request.getCode()) {
- case RequestCode.SEND_BATCH_MESSAGE:
- case RequestCode.SEND_MESSAGE_V2:
- requestHeaderV2 = request.decodeCommandCustomHeader(SendMessageRequestHeaderV2.class);
- case RequestCode.SEND_MESSAGE:
- if (null == requestHeaderV2) {
- requestHeader = request.decodeCommandCustomHeader(SendMessageRequestHeader.class);
- } else {
- requestHeader = SendMessageRequestHeaderV2.createSendMessageRequestHeaderV1(requestHeaderV2);
- }
- default:
- break;
- }
- return requestHeader;
+ return request.decodeCommandCustomHeader(SendMessageRequestHeader.class);
+ }
+
+ @Override
+ public void encode(ByteBuf out) {
+ writeIfNotNull(out, "a", producerGroup);
+ writeIfNotNull(out, "b", topic);
+ writeIfNotNull(out, "c", defaultTopic);
+ writeIfNotNull(out, "d", defaultTopicQueueNums);
+ writeIfNotNull(out, "e", queueId);
+ writeIfNotNull(out, "f", sysFlag);
+ writeIfNotNull(out, "g", bornTimestamp);
+ writeIfNotNull(out, "h", flag);
+ writeIfNotNull(out, "i", properties);
+ writeIfNotNull(out, "j", reconsumeTimes);
+ writeIfNotNull(out, "k", unitMode);
+ writeIfNotNull(out, "l", maxReconsumeTimes);
+ writeIfNotNull(out, "m", batch);
+ writeIfNotNull(out, "n", getBname());
+ }
+
+ @Override
+ public void decode(HashMap<String, String> fields) throws RemotingCommandException {
+ String str = fields.get("a");
+ if (str == null) {
+ str = getAndCheckNotNull(fields, "producerGroup");
+ }
+ if (str != null) {
+ this.producerGroup = str;
+ }
+
+ str = fields.get("b");
+ if (str == null) {
+ str = getAndCheckNotNull(fields, "topic");
+ }
+ if (str != null) {
+ this.topic = str;
+ }
+
+ str = fields.get("c");
+ if (str == null) {
+ str = getAndCheckNotNull(fields, "defaultTopic");
+ }
+ if (str != null) {
+ this.defaultTopic = str;
+ }
+
+ str = fields.get("d");
+ if (str == null) {
+ str = getAndCheckNotNull(fields, "defaultTopicQueueNums");
+ }
+ if (str != null) {
+ this.defaultTopicQueueNums = Integer.parseInt(str);
+ }
+
+ str = fields.get("e");
+ if (str == null) {
+ str = getAndCheckNotNull(fields, "queueId");
+ }
+ if (str != null) {
+ this.queueId = Integer.parseInt(str);
+ }
+
+ str = fields.get("f");
+ if (str == null) {
+ str = getAndCheckNotNull(fields, "sysFlag");
+ }
+ if (str != null) {
+ this.sysFlag = Integer.parseInt(str);
+ }
+
+ str = fields.get("g");
+ if (str == null) {
+ str = getAndCheckNotNull(fields, "bornTimestamp");
+ }
+ if (str != null) {
+ this.bornTimestamp = Long.parseLong(str);
+ }
+
+ str = fields.get("h");
+ if (str == null) {
+ str = getAndCheckNotNull(fields, "flag");
+ }
+ if (str != null) {
+ this.flag = Integer.parseInt(str);
+ }
+
+ str = fields.get("i");
+ if (str == null) {
+ str = fields.get("properties");
+ }
+ if (str != null) {
+ this.properties = str;
+ }
+
+ str = fields.get("j");
+ if (str == null) {
+ str = fields.get("reconsumeTimes");
+ }
+ if (str != null) {
+ this.reconsumeTimes = Integer.parseInt(str);
+ }
+
+ str = fields.get("k");
+ if (str == null) {
+ str = fields.get("unitMode");
+ }
+ if (str != null) {
+ this.unitMode = Boolean.parseBoolean(str);
+ }
+
+ str = fields.get("l");
+ if (str == null) {
+ str = fields.get("maxReconsumeTimes");
+ }
+ if (str != null) {
+ this.maxReconsumeTimes = Integer.parseInt(str);
+ }
+
+ str = fields.get("m");
+ if (str == null) {
+ str = fields.get("batch");
+ }
+ if (str != null) {
+ this.batch = Boolean.parseBoolean(str);
+ }
+
+ str = fields.get("n");
+ if (str == null) {
+ str = fields.get("brokerName");
+ }
+ if (str != null) {
+ this.setBrokerName(str);
+ }
Review Comment:
decode() falls back from short key "n" only to "brokerName", but older V1 reflection-based encoding (makeCustomHeaderToNet) serializes the inherited RpcRequestHeader field as "bname". After switching parseRequestHeader() to always decode via SendMessageRequestHeader.decode(), brokerName may be silently dropped for mixed-version traffic.
Add a fallback for "bname" when decoding broker name.
##########
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/SendMessageRequestHeader.java:
##########
@@ -183,22 +190,140 @@ public void setBatch(Boolean batch) {
}
public static SendMessageRequestHeader parseRequestHeader(RemotingCommand request) throws RemotingCommandException {
- SendMessageRequestHeaderV2 requestHeaderV2 = null;
- SendMessageRequestHeader requestHeader = null;
- switch (request.getCode()) {
- case RequestCode.SEND_BATCH_MESSAGE:
- case RequestCode.SEND_MESSAGE_V2:
- requestHeaderV2 = request.decodeCommandCustomHeader(SendMessageRequestHeaderV2.class);
- case RequestCode.SEND_MESSAGE:
- if (null == requestHeaderV2) {
- requestHeader = request.decodeCommandCustomHeader(SendMessageRequestHeader.class);
- } else {
- requestHeader = SendMessageRequestHeaderV2.createSendMessageRequestHeaderV1(requestHeaderV2);
- }
- default:
- break;
- }
- return requestHeader;
+ return request.decodeCommandCustomHeader(SendMessageRequestHeader.class);
+ }
+
+ @Override
Review Comment:
SendMessageRequestHeader is now a FastCodesHeader with custom encode/decode and a new mixed-key decode strategy (short "a".."n" + long names). There is existing test coverage for SendMessageRequestHeaderV2, but no tests exercising SendMessageRequestHeader’s new encode/decode behavior or V1/V2 compatibility.
Add unit tests to validate: (1) encoding includes expected keys for SEND_MESSAGE compatibility, and (2) decoding works with both legacy long keys and short keys (including brokerName/bname fallback).
--
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: [PR] [ISSUE #10538] Eliminate SendMessageRequestHeaderV2 intermediate object in client [rocketmq]
Posted by "wang-jiahua (via GitHub)" <gi...@apache.org>.
wang-jiahua commented on PR #10540:
URL: https://github.com/apache/rocketmq/pull/10540#issuecomment-4775103758
## Server Benchmark Results (2026-06-23)
**Environment**: 8C30G broker, Dragonwell JDK 21, 1KB msg, 128 queues, sync mode, 256 threads, 20s warmup + 30s collect
### Producer-only (steady state, lines 3-5)
| Test | TPS | Avg RT | GC | vs Baseline |
|------|-----|--------|----|-------------|
| Baseline | 217k | 1.18ms | 354 | — |
| #10540 | 207k | 1.23ms | 238 | -4.6% TPS (noise), **-33% GC** ✅ |
Client skip V2 intermediate object eliminates per-send SendMessageRequestHeaderV2 allocation + 14 field copies. TPS within noise band. GC reduction (-33%) confirms allocation savings on client side.
--
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: [PR] [ISSUE #10538] Eliminate SendMessageRequestHeaderV2 intermediate object in client [rocketmq]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #10540:
URL: https://github.com/apache/rocketmq/pull/10540#issuecomment-4787348748
## [Codecov](https://app.codecov.io/gh/apache/rocketmq/pull/10540?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `1.07527%` with `92 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 48.12%. Comparing base ([`2e6632f`](https://app.codecov.io/gh/apache/rocketmq/commit/2e6632ff404b6210057914cf0530f53d6531ab99?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`885f563`](https://app.codecov.io/gh/apache/rocketmq/commit/885f5638247567c7d89bac4bad9ab439cc211960?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/rocketmq/pull/10540?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...ting/protocol/header/SendMessageRequestHeader.java](https://app.codecov.io/gh/apache/rocketmq/pull/10540?src=pr&el=tree&filepath=remoting%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fremoting%2Fprotocol%2Fheader%2FSendMessageRequestHeader.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cmVtb3Rpbmcvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3JlbW90aW5nL3Byb3RvY29sL2hlYWRlci9TZW5kTWVzc2FnZVJlcXVlc3RIZWFkZXIuamF2YQ==) | 0.00% | [89 Missing :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/10540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...g/apache/rocketmq/client/impl/MQClientAPIImpl.java](https://app.codecov.io/gh/apache/rocketmq/pull/10540?src=pr&el=tree&filepath=client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fclient%2Fimpl%2FMQClientAPIImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jbGllbnQvaW1wbC9NUUNsaWVudEFQSUltcGwuamF2YQ==) | 0.00% | [0 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/10540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../rocketmq/client/impl/mqclient/MQClientAPIExt.java](https://app.codecov.io/gh/apache/rocketmq/pull/10540?src=pr&el=tree&filepath=client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fclient%2Fimpl%2Fmqclient%2FMQClientAPIExt.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jbGllbnQvaW1wbC9tcWNsaWVudC9NUUNsaWVudEFQSUV4dC5qYXZh) | 50.00% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/10540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## develop #10540 +/- ##
=============================================
- Coverage 48.26% 48.12% -0.15%
+ Complexity 13433 13414 -19
=============================================
Files 1377 1377
Lines 100844 100917 +73
Branches 13036 13062 +26
=============================================
- Hits 48672 48563 -109
- Misses 46221 46387 +166
- Partials 5951 5967 +16
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/rocketmq/pull/10540?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
--
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