AsperforMias commented on code in PR #1125:
URL:
https://github.com/apache/incubator-seata-go/pull/1125#discussion_r3485975087
##########
pkg/integration/rocketmq/tcc_rocketmq_action.go:
##########
@@ -78,27 +86,98 @@ func (a *TCCRocketMQAction) Prepare(ctx context.Context,
params interface{}) (bo
bac.ActionContext[ActionContextKeyQueueId] =
result.MessageQueue.QueueId
bac.ActionContext[ActionContextKeyBrokerName] =
result.MessageQueue.BrokerName
}
+ bac.ActionContext[ActionContextKeyTopic] = msg.Topic
log.Infof("[TCCRocketMQ] Prepare success, xid=%s, branchId=%d,
msgId=%s", xid, bac.BranchId, result.MsgID)
return true, nil
}
func (a *TCCRocketMQAction) Commit(ctx context.Context, bac
*tm.BusinessActionContext) (bool, error) {
- // Commit is a no-op because RocketMQ transactional messages use a
check-back mechanism.
- // When the global transaction commits, RocketMQ will invoke
CheckLocalTransaction
- // via SeataTransactionListener to determine the final message
disposition.
- // The message has already been sent to the broker during Prepare phase
with an
- // initial state of UnknowState, pending the check-back resolution.
- log.Infof("[TCCRocketMQ] Commit (no-op, rely on check-back), xid=%s,
branchId=%d", bac.Xid, bac.BranchId)
+ if a.producer == nil || a.producer.config == nil {
+ log.Warnf("[TCCRocketMQ] Commit skipped, producer or config is
nil, fallback to check-back, xid=%s, branchId=%d", bac.Xid, bac.BranchId)
+ return true, nil
+ }
+ topic := getStringFromMap(bac.ActionContext, ActionContextKeyTopic)
+ brokerName := getStringFromMap(bac.ActionContext,
ActionContextKeyBrokerName)
+ if topic == "" || brokerName == "" {
+ log.Warnf("[TCCRocketMQ] Commit missing metadata (topic=%s,
brokerName=%s), skip active END_TRANSACTION, fallback to check-back, xid=%s,
branchId=%d",
+ topic, brokerName, bac.Xid, bac.BranchId)
+ return true, nil
+ }
+ header := a.buildEndTransactionHeader(bac, topic,
commitOrRollbackCommit)
+ err := sendEndTransaction(
+ a.producer.config.NameServerAddrs,
+ topic,
+ brokerName,
+ header,
+ a.producer.config.SendMsgTimeout,
+ a.resolver,
+ a.sender,
+ )
+ if err != nil {
+ log.Warnf("[TCCRocketMQ] Commit send END_TRANSACTION failed,
fallback to check-back, xid=%s, branchId=%d, err=%v",
+ bac.Xid, bac.BranchId, err)
+ return true, nil
+ }
+ log.Infof("[TCCRocketMQ] Commit send END_TRANSACTION success, xid=%s,
branchId=%d", bac.Xid, bac.BranchId)
return true, nil
}
func (a *TCCRocketMQAction) Rollback(ctx context.Context, bac
*tm.BusinessActionContext) (bool, error) {
- // Rollback is a no-op because RocketMQ transactional messages use a
check-back mechanism.
- // When the global transaction rolls back, RocketMQ will invoke
CheckLocalTransaction
- // via SeataTransactionListener, which queries the TC for the global
status and returns
- // RollbackMessageState, causing the broker to discard the message.
- log.Infof("[TCCRocketMQ] Rollback (no-op, rely on check-back), xid=%s,
branchId=%d", bac.Xid, bac.BranchId)
+ if a.producer == nil || a.producer.config == nil {
+ log.Warnf("[TCCRocketMQ] Rollback skipped, producer or config
is nil, fallback to check-back, xid=%s, branchId=%d", bac.Xid, bac.BranchId)
+ return true, nil
+ }
+ topic := getStringFromMap(bac.ActionContext, ActionContextKeyTopic)
+ brokerName := getStringFromMap(bac.ActionContext,
ActionContextKeyBrokerName)
+ if topic == "" || brokerName == "" {
+ log.Warnf("[TCCRocketMQ] Rollback missing metadata (topic=%s,
brokerName=%s), skip active END_TRANSACTION, fallback to check-back, xid=%s,
branchId=%d",
+ topic, brokerName, bac.Xid, bac.BranchId)
+ return true, nil
+ }
+ header := a.buildEndTransactionHeader(bac, topic,
commitOrRollbackRollback)
+ err := sendEndTransaction(
+ a.producer.config.NameServerAddrs,
+ topic,
+ brokerName,
+ header,
+ a.producer.config.SendMsgTimeout,
+ a.resolver,
+ a.sender,
+ )
+ if err != nil {
+ log.Warnf("[TCCRocketMQ] Rollback send END_TRANSACTION failed,
fallback to check-back, xid=%s, branchId=%d, err=%v",
+ bac.Xid, bac.BranchId, err)
+ return true, nil
+ }
+ log.Infof("[TCCRocketMQ] Rollback send END_TRANSACTION success, xid=%s,
branchId=%d", bac.Xid, bac.BranchId)
return true, nil
}
+
+func (a *TCCRocketMQAction) buildEndTransactionHeader(bac
*tm.BusinessActionContext, topic string, commitOrRollback int)
*endTransactionRequestHeader {
+ actionCtx := bac.ActionContext
+ if actionCtx == nil {
+ actionCtx = make(map[string]interface{})
+ }
+
+ offsetMsgID := getStringFromMap(actionCtx, ActionContextKeyOffsetMsgId)
Review Comment:
offsetMsgId 解析失败时，感觉不应该继续发 0 offset
这里 offsetMsgId 解析失败时会把 commitLogOffset 默认为 0，然后继续发
END_TRANSACTION。这个感觉有点怪，因为broker 可能会按错误 offset 处理半消息。官方 client 的 endTransaction
是 offsetMsgId 为空时 fallback 到 msgId。这里要么也做 fallback，要么解析失败时直接跳过主动通知，走 check-back
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
