feat:support rocketmq mode in tcc#1125
Conversation
github-actions
Bot
added
coding
rm
integration
enhancement
There was a problem hiding this comment.
Pull request overview
This PR adds active RocketMQ transactional message finalization for Seata TCC by sending END_TRANSACTION directly to the broker (commit/rollback) and by ensuring the TCC phase-two has enough message metadata via delayed ActionContext reporting to the TC. It also adjusts RocketMQ check-back status mapping to avoid premature rollback when the global status is CommitFailed.
Changes:
- Add an
END_TRANSACTIONremoting implementation (NameServer route query + TCP sender) and switch TCC RocketMQ phase-two from no-op to best-effort broker notification with check-back fallback. - Delay-report updated
ActionContextafter a successful TCC Prepare so phase-two can access RocketMQ message/broker metadata. - Fix
GlobalStatusCommitFailedmapping toUnknownStateand update related tests; add RocketMQ producer defaultSendMsgTimeout.
Reviewed changes
Copilot reviewed 9 out of 9 changed files in this pull request and generated 8 comments.
Show a summary per file
| File | Description |
|---|---|
| pkg/rm/tcc/tcc_service.go | Report updated ActionContext after Prepare (delayed report). |
| pkg/integration/rocketmq/transaction_listener.go | Adjust global-status → RocketMQ transaction-state mapping. |
| pkg/integration/rocketmq/transaction_listener_test.go | Update mapping expectations. |
| pkg/integration/rocketmq/tcc_rocketmq_action.go | Actively send END_TRANSACTION on Commit/Rollback; build request header from ActionContext. |
| pkg/integration/rocketmq/tcc_rocketmq_action_test.go | Add tests for active Commit/Rollback and header building. |
| pkg/integration/rocketmq/seata_producer.go | Add default SendMsgTimeout configuration. |
| pkg/integration/rocketmq/end_transaction_sender.go | New TCP remoting implementation for NameServer route lookup + broker END_TRANSACTION. |
| pkg/integration/rocketmq/end_transaction_sender_test.go | Add unit tests for protocol encoding and helper logic. |
| pkg/integration/rocketmq/constants.go | Add topic ActionContext key. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #1125 +/- ##
==========================================
- Coverage 58.36% 58.10% -0.27%
==========================================
Files 282 283 +1
Lines 19405 19750 +345
==========================================
+ Hits 11326 11475 +149
- Misses 7102 7272 +170
- Partials 977 1003 +26 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
新增修改 pkg/integration/rocketmq/end_transaction_sender.go
|
Add Amendment pkg/integration/rocketmq/end_transaction_sender.go
|
| actionCtx = make(map[string]interface{}) | ||
| } | ||
|
|
||
| offsetMsgID := getStringFromMap(actionCtx, ActionContextKeyOffsetMsgId) |
There was a problem hiding this comment.
offsetMsgId 解析失败时，感觉不应该继续发 0 offset
这里 offsetMsgId 解析失败时会把 commitLogOffset 默认为 0，然后继续发 END_TRANSACTION。这个感觉有点怪，因为broker 可能会按错误 offset 处理半消息。官方 client 的 endTransaction 是 offsetMsgId 为空时 fallback 到 msgId。这里要么也做 fallback，要么解析失败时直接跳过主动通知，走 check-back
|
新增修改 pkg/integration/rocketmq/tcc_rocketmq_action.go 问题: offsetMsgId 解析失败时 commitLogOffset 默认为 0，继续发送 END_TRANSACTION 可能导致 broker 按错误 offset 处理半消息 |
Add Amendment pkg/integration/rocketmq/tcc_rocketmq_action.go Problem: when offsetMsgId parsing fails, commitLogOffset defaults to 0, continuing to send end_transaction may cause broker to process half messages by error offset |
| } | ||
|
|
||
| // fallback: if offsetMsgId is empty or failed to parse, try msgId (aligned with official client behaviour) | ||
| if commitLogOffset == 0 { |
There was a problem hiding this comment.
这里比之前好一些，但如果 offsetMsgId 和 msgId 都缺失或解析失败，commitLogOffset 还是会保持 0 并继续发送 END_TRANSACTION。这个场景下可能应该直接跳过主动通知，走 check-back？否则到最后还是可能把无效 offset 发给 broker
| log.Errorf("[TCC] marshal updated ActionContext failed, xid=%s, branchId=%d, err=%v", bac.Xid, bac.BranchId, err) | ||
| return fmt.Errorf("marshal updated ActionContext failed: %w", err) | ||
| } | ||
| err = rm.GetRMRemotingInstance().BranchReport(rm.BranchReportParam{ |
There was a problem hiding this comment.
这里依赖 BranchReport 把更新后的 ActionContext 写回 TC。能确认 TC 会更新已注册分支的 applicationData 吗？如果这里只是状态上报，那第二阶段拿到的还是旧 applicationData，主动 END_TRANSACTION 就会退回 check-back。可以补个测试或者大概说明一下这个依赖
What this PR does:
This PR integrates the
END_TRANSACTIONprotocol adaptation into the TCC RocketMQ mode, enabling immediate Commit/Rollback notifications to the Broker and eliminating the reliance on the 60-second checkback mechanism.END_TRANSACTIONrequest (RequestCode=37) and sends Commit/Rollback instructions directly to the Broker via TCP.RequestCode=105) to automatically resolve the Broker master node address.Commit(): Constructs theEND_TRANSACTIONrequest (commitOrRollback=8) and notifies the Broker to commit the half message viasendEndTransaction(). On failure, it logs a warning, returnstrue, and falls back to the Broker checkback.Rollback(): Constructs theEND_TRANSACTIONrequest (commitOrRollback=12) and notifies the Broker to roll back the half message viasendEndTransaction(). On failure, it similarly falls back to the checkback.buildEndTransactionHeader(): Extracts metadata such asoffsetMsgId,brokerName,topic, andqueueOffsetfromActionContext, and parsesCommitLogOffset.GlobalStatusCommitFailedto returnUnknownState, preventing the Broker from directly rolling back during checkback and allowing the TC to continue commit retries.ActionContextmsgId,brokerName,topic, etc.) back to the TC viaBranchReport, ensuring the phase-two Commit/Rollback can access the Broker address information.Which issue(s) this PR fixes?
Related to #765
Special notes for your reviewer:
Does this PR introduce a user-facing change?: