This is an automated email from the ASF dual-hosted git repository.
thunguo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-seata-go.git
The following commit(s) were added to refs/heads/master by this push:
new 0018e93e feat:support rocketmq mode in tcc (#1125)
0018e93e is described below
commit 0018e93ee23620431c8fc7c3f703c460872088b8
Author: XiaoFei <[email protected]>
AuthorDate: Fri Jul 10 21:17:55 2026 +0800
feat:support rocketmq mode in tcc (#1125)
* feat:support rocketmq mode in tcc
* fix: make reportActionContext failure non-fatal in TCC Prepare
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
* fix: guard Commit/Rollback against missing metadata
* fix: end_transaction_sender frame parsing hardening against OOM/panic
* fix: address PR review comments
* fix: fallback to msgId when offsetMsgId parsing fails
* fix: fallback to msgId when offsetMsgId fails; if both missing/invalid,
skip active END_TRANSACTION and fallback to check-back
* fix: fix golangci-lint misspell errors
* fix: fix CI problems
* fix: fix CI problems
---------
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
Co-authored-by: ThunGuo <[email protected]>
---
pkg/integration/rocketmq/config.go | 2 +
pkg/integration/rocketmq/constants.go | 1 +
pkg/integration/rocketmq/end_transaction_sender.go | 628 ++++++++++++++++++++
.../rocketmq/end_transaction_sender_test.go | 652 +++++++++++++++++++++
pkg/integration/rocketmq/seata_producer.go | 13 +
pkg/integration/rocketmq/tcc_rocketmq_action.go | 121 +++-
.../rocketmq/tcc_rocketmq_action_test.go | 311 ++++++++++
pkg/integration/rocketmq/transaction_listener.go | 4 +-
.../rocketmq/transaction_listener_test.go | 2 +-
pkg/rm/tcc/tcc_service.go | 48 +-
pkg/rm/tcc/tcc_service_test.go | 55 ++
pkg/util/flagext/day_test.go | 33 +-
12 files changed, 1823 insertions(+), 47 deletions(-)
diff --git a/pkg/integration/rocketmq/config.go
b/pkg/integration/rocketmq/config.go
index e97e3b36..ab365151 100644
--- a/pkg/integration/rocketmq/config.go
+++ b/pkg/integration/rocketmq/config.go
@@ -31,12 +31,14 @@ type SeataMQProducerConfig struct {
RetryTimesWhenSendFailed int
SendMsgTimeout time.Duration
+ ConnPoolSize int
}
func NewDefaultSeataMQProducerConfig() *SeataMQProducerConfig {
return &SeataMQProducerConfig{
RetryTimesWhenSendFailed: 3,
SendMsgTimeout: 3 * time.Second,
+ ConnPoolSize: 4,
}
}
diff --git a/pkg/integration/rocketmq/constants.go
b/pkg/integration/rocketmq/constants.go
index b0bbbe7f..7e440ce3 100644
--- a/pkg/integration/rocketmq/constants.go
+++ b/pkg/integration/rocketmq/constants.go
@@ -26,4 +26,5 @@ const (
ActionContextKeyQueueOffset = "queueOffset"
ActionContextKeyTransactionId = "transactionId"
ActionContextKeyBrokerName = "brokerName"
+ ActionContextKeyTopic = "topic"
)
diff --git a/pkg/integration/rocketmq/end_transaction_sender.go
b/pkg/integration/rocketmq/end_transaction_sender.go
new file mode 100644
index 00000000..6e80cfcd
--- /dev/null
+++ b/pkg/integration/rocketmq/end_transaction_sender.go
@@ -0,0 +1,628 @@
+/*
+ * 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.
+ */
+
+package rocketmq
+
+import (
+ "bytes"
+ "encoding/binary"
+ "encoding/json"
+ "fmt"
+ "io"
+ "net"
+ "sync"
+ "sync/atomic"
+ "time"
+)
+
+const (
+ reqEndTransaction int16 = 37
+
+ reqGetRouteInfoByTopic int16 = 105
+
+ rmqProtocolVersion int16 = 317
+
+ rmqLanguageGo byte = 9
+
+ rmqCodecType byte = 1
+
+ rmqHeaderFixedLength = 21
+
+ // rmqResponseHeaderFixedPartLength is the size of the fixed fields in a
+ // RocketMQ remoting response header:
+ // code(2) + language(1) + version(2) + opaque(4) + flag(4) = 13 bytes.
+ rmqResponseHeaderFixedPartLength = 13
+
+ commitOrRollbackCommit = 8
+
+ commitOrRollbackRollback = 12
+
+ maxFrameSize = 1024 * 1024 // 1MB upper bound for remoting frame to
prevent OOM
+
+ defaultConnPoolSize = 4
+
+ connPoolCleanInterval = 60 // seconds
+
+ connPoolIdleTimeout = 600 // seconds (10 minutes)
+)
+
+var opaqueCounter int32
+
+type endTransactionRequestHeader struct {
+ Topic string
+ ProducerGroup string
+ TranStateTableOffset int64
+ CommitLogOffset int64
+ CommitOrRollback int
+ FromTransactionCheck bool
+ MsgID string
+ TransactionId string
+}
+
+func (h *endTransactionRequestHeader) Encode() map[string]string {
+ return map[string]string{
+ "topic": h.Topic,
+ "producerGroup": h.ProducerGroup,
+ "tranStateTableOffset": fmt.Sprintf("%d",
h.TranStateTableOffset),
+ "commitLogOffset": fmt.Sprintf("%d", h.CommitLogOffset),
+ "commitOrRollback": fmt.Sprintf("%d", h.CommitOrRollback),
+ "fromTransactionCheck": fmt.Sprintf("%v",
h.FromTransactionCheck),
+ "msgId": h.MsgID,
+ "transactionId": h.TransactionId,
+ }
+}
+
+type remotingCommand struct {
+ Code int16
+ Language byte
+ Version int16
+ Opaque int32
+ Flag int32
+ Remark string
+ ExtFields map[string]string
+ Body []byte
+}
+
+func newEndTransactionCommand(header *endTransactionRequestHeader)
*remotingCommand {
+ return &remotingCommand{
+ Code: reqEndTransaction,
+ Language: rmqLanguageGo,
+ Version: rmqProtocolVersion,
+ Opaque: atomic.AddInt32(&opaqueCounter, 1),
+ ExtFields: header.Encode(),
+ }
+}
+
+func (cmd *remotingCommand) encode() ([]byte, error) {
+ headerBytes, err := cmd.encodeHeader()
+ if err != nil {
+ return nil, fmt.Errorf("encode header failed: %w", err)
+ }
+
+ frameSize := 4 + len(headerBytes) + len(cmd.Body)
+ buf := bytes.NewBuffer(make([]byte, 0, 4+frameSize))
+
+ if err := binary.Write(buf, binary.BigEndian, int32(frameSize)); err !=
nil {
+ return nil, err
+ }
+ if err := binary.Write(buf, binary.BigEndian,
markProtocolType(int32(len(headerBytes)))); err != nil {
+ return nil, err
+ }
+ if _, err := buf.Write(headerBytes); err != nil {
+ return nil, err
+ }
+ if len(cmd.Body) > 0 {
+ if _, err := buf.Write(cmd.Body); err != nil {
+ return nil, err
+ }
+ }
+
+ return buf.Bytes(), nil
+}
+
+func (cmd *remotingCommand) encodeHeader() ([]byte, error) {
+ extBytes, err := encodeExtFields(cmd.ExtFields)
+ if err != nil {
+ return nil, err
+ }
+
+ buf := bytes.NewBuffer(make([]byte, 0,
rmqHeaderFixedLength+len(cmd.Remark)+len(extBytes)))
+
+ if err := binary.Write(buf, binary.BigEndian, cmd.Code); err != nil {
+ return nil, err
+ }
+ if err := buf.WriteByte(cmd.Language); err != nil {
+ return nil, err
+ }
+ if err := binary.Write(buf, binary.BigEndian, cmd.Version); err != nil {
+ return nil, err
+ }
+ if err := binary.Write(buf, binary.BigEndian, cmd.Opaque); err != nil {
+ return nil, err
+ }
+ if err := binary.Write(buf, binary.BigEndian, cmd.Flag); err != nil {
+ return nil, err
+ }
+ if err := binary.Write(buf, binary.BigEndian, int32(len(cmd.Remark)));
err != nil {
+ return nil, err
+ }
+ if len(cmd.Remark) > 0 {
+ if _, err := buf.Write([]byte(cmd.Remark)); err != nil {
+ return nil, err
+ }
+ }
+ if err := binary.Write(buf, binary.BigEndian, int32(len(extBytes)));
err != nil {
+ return nil, err
+ }
+ if len(extBytes) > 0 {
+ if _, err := buf.Write(extBytes); err != nil {
+ return nil, err
+ }
+ }
+
+ return buf.Bytes(), nil
+}
+
+func encodeExtFields(fields map[string]string) ([]byte, error) {
+ if len(fields) == 0 {
+ return []byte{}, nil
+ }
+
+ buf := bytes.NewBuffer(nil)
+ for key, value := range fields {
+ if err := binary.Write(buf, binary.BigEndian, int16(len(key)));
err != nil {
+ return nil, err
+ }
+ if _, err := buf.Write([]byte(key)); err != nil {
+ return nil, err
+ }
+ if err := binary.Write(buf, binary.BigEndian,
int32(len(value))); err != nil {
+ return nil, err
+ }
+ if _, err := buf.Write([]byte(value)); err != nil {
+ return nil, err
+ }
+ }
+ return buf.Bytes(), nil
+}
+
+func markProtocolType(source int32) []byte {
+ result := make([]byte, 4)
+ result[0] = rmqCodecType
+ result[1] = byte((source >> 16) & 0xFF)
+ result[2] = byte((source >> 8) & 0xFF)
+ result[3] = byte(source & 0xFF)
+ return result
+}
+
+type brokerAddrResolver interface {
+ ResolveBrokerAddr(nameServerAddrs []string, topic string, brokerName
string, timeout time.Duration) (string, error)
+}
+
+type tcpSender interface {
+ Send(addr string, data []byte, timeout time.Duration) error
+ Close() error
+}
+
+type defaultBrokerAddrResolver struct{}
+
+func (r *defaultBrokerAddrResolver) ResolveBrokerAddr(nameServerAddrs
[]string, topic string, brokerName string, timeout time.Duration) (string,
error) {
+ for _, nsAddr := range nameServerAddrs {
+ addr, err := queryBrokerAddrFromNameServer(nsAddr, topic,
brokerName, timeout)
+ if err == nil && addr != "" {
+ return addr, nil
+ }
+ }
+ return "", fmt.Errorf("broker %s addr not found from name servers",
brokerName)
+}
+
+type connPool struct {
+ addr string
+ conns chan net.Conn
+ lastUsedAt int64 // atomic, unix timestamp
+ closed int32 // atomic
+}
+
+func newConnPool(addr string, size int) *connPool {
+ if size <= 0 {
+ size = defaultConnPoolSize
+ }
+ return &connPool{
+ addr: addr,
+ conns: make(chan net.Conn, size),
+ lastUsedAt: time.Now().Unix(),
+ }
+}
+
+// get returns a connection and a bool indicating whether it was freshly
+// dialed (true) or reused from the pool (false).
+func (p *connPool) get(timeout time.Duration) (net.Conn, bool, error) {
+ atomic.StoreInt64(&p.lastUsedAt, time.Now().Unix())
+ select {
+ case conn := <-p.conns:
+ return conn, false, nil
+ default:
+ }
+ conn, err := net.DialTimeout("tcp", p.addr, timeout)
+ return conn, true, err
+}
+
+func (p *connPool) put(conn net.Conn) {
+ if atomic.LoadInt32(&p.closed) == 1 {
+ conn.Close()
+ return
+ }
+ select {
+ case p.conns <- conn:
+ default:
+ conn.Close()
+ }
+}
+
+func (p *connPool) closeAll() {
+ atomic.StoreInt32(&p.closed, 1)
+ for {
+ select {
+ case conn := <-p.conns:
+ conn.Close()
+ default:
+ return
+ }
+ }
+}
+
+type defaultTCPSender struct {
+ pools sync.Map // addr -> *connPool
+ poolSize int
+ lastCleanedAt int64 // atomic, unix timestamp
+}
+
+func newDefaultTCPSender(poolSize int) *defaultTCPSender {
+ if poolSize <= 0 {
+ poolSize = defaultConnPoolSize
+ }
+ return &defaultTCPSender{
+ poolSize: poolSize,
+ lastCleanedAt: time.Now().Unix(),
+ }
+}
+
+func (s *defaultTCPSender) Send(addr string, data []byte, timeout
time.Duration) error {
+ s.cleanIdlePoolsIfNeeded()
+
+ poolAny, _ := s.pools.LoadOrStore(addr, newConnPool(addr, s.poolSize))
+ pool := poolAny.(*connPool)
+
+ conn, isNew, err := pool.get(timeout)
+ if err != nil {
+ return fmt.Errorf("dial broker %s failed: %w", addr, err)
+ }
+
+ sendAndRead := func(c net.Conn) error {
+ if err := c.SetWriteDeadline(time.Now().Add(timeout)); err !=
nil {
+ return fmt.Errorf("set write deadline failed: %w", err)
+ }
+ if _, err := c.Write(data); err != nil {
+ return fmt.Errorf("write to broker %s failed: %w",
addr, err)
+ }
+
+ if err := c.SetReadDeadline(time.Now().Add(timeout)); err !=
nil {
+ return fmt.Errorf("set read deadline failed: %w", err)
+ }
+ frameLenBuf := make([]byte, 4)
+ if _, err := io.ReadFull(c, frameLenBuf); err != nil {
+ return fmt.Errorf("read response frame length from
broker %s failed: %w", addr, err)
+ }
+ frameLen := int(binary.BigEndian.Uint32(frameLenBuf))
+ if frameLen < 8 {
+ return fmt.Errorf("broker %s response frame too short:
%d", addr, frameLen)
+ }
+ if frameLen > maxFrameSize {
+ return fmt.Errorf("broker %s response frame too large:
%d", addr, frameLen)
+ }
+ frameBuf := make([]byte, frameLen)
+ if _, err := io.ReadFull(c, frameBuf); err != nil {
+ return fmt.Errorf("read response frame from broker %s
failed: %w", addr, err)
+ }
+
+ code := int16(binary.BigEndian.Uint16(frameBuf[4:6]))
+ if code != 0 {
+ return fmt.Errorf("broker %s rejected END_TRANSACTION,
responseCode=%d", addr, code)
+ }
+ return nil
+ }
+
+ if err := sendAndRead(conn); err != nil {
+ conn.Close()
+ // If the connection was pooled (not freshly dialed), it may
have
+ // been closed by the broker during idle time. Retry once with a
+ // fresh connection before giving up.
+ if !isNew {
+ freshConn, dialErr := net.DialTimeout("tcp", addr,
timeout)
+ if dialErr == nil {
+ if retryErr := sendAndRead(freshConn); retryErr
!= nil {
+ freshConn.Close()
+ return retryErr
+ }
+ pool.put(freshConn)
+ return nil
+ }
+ }
+ return err
+ }
+
+ pool.put(conn)
+ return nil
+}
+
+func (s *defaultTCPSender) Close() error {
+ s.pools.Range(func(key, value interface{}) bool {
+ pool := value.(*connPool)
+ if _, loaded := s.pools.LoadAndDelete(key); loaded {
+ pool.closeAll()
+ }
+ return true
+ })
+ return nil
+}
+
+func (s *defaultTCPSender) cleanIdlePoolsIfNeeded() {
+ now := time.Now().Unix()
+ lastClean := atomic.LoadInt64(&s.lastCleanedAt)
+ if now-lastClean < int64(connPoolCleanInterval) {
+ return
+ }
+ if !atomic.CompareAndSwapInt64(&s.lastCleanedAt, lastClean, now) {
+ return
+ }
+
+ s.pools.Range(func(key, value interface{}) bool {
+ pool := value.(*connPool)
+ if now-atomic.LoadInt64(&pool.lastUsedAt) >
int64(connPoolIdleTimeout) {
+ if _, loaded := s.pools.LoadAndDelete(key); loaded {
+ pool.closeAll()
+ }
+ }
+ return true
+ })
+}
+
+func sendEndTransaction(
+ nameServerAddrs []string,
+ topic string,
+ brokerName string,
+ header *endTransactionRequestHeader,
+ timeout time.Duration,
+ resolver brokerAddrResolver,
+ sender tcpSender,
+) error {
+ brokerAddr, err := resolver.ResolveBrokerAddr(nameServerAddrs, topic,
brokerName, timeout)
+ if err != nil {
+ return fmt.Errorf("resolve broker addr failed: %w", err)
+ }
+
+ cmd := newEndTransactionCommand(header)
+
+ data, err := cmd.encode()
+ if err != nil {
+ return fmt.Errorf("encode command failed: %w", err)
+ }
+
+ if err := sender.Send(brokerAddr, data, timeout); err != nil {
+ return fmt.Errorf("send to broker %s failed: %w", brokerAddr,
err)
+ }
+
+ return nil
+}
+
+func queryBrokerAddrFromNameServer(nameServerAddr string, topic string,
brokerName string, timeout time.Duration) (string, error) {
+ cmd := &remotingCommand{
+ Code: reqGetRouteInfoByTopic,
+ Language: rmqLanguageGo,
+ Version: rmqProtocolVersion,
+ Opaque: atomic.AddInt32(&opaqueCounter, 1),
+ ExtFields: map[string]string{
+ "topic": topic,
+ },
+ }
+
+ data, err := cmd.encode()
+ if err != nil {
+ return "", fmt.Errorf("encode route request failed: %w", err)
+ }
+
+ conn, err := net.DialTimeout("tcp", nameServerAddr, timeout)
+ if err != nil {
+ return "", fmt.Errorf("dial name server %s failed: %w",
nameServerAddr, err)
+ }
+ defer conn.Close()
+
+ if err := conn.SetWriteDeadline(time.Now().Add(timeout)); err != nil {
+ return "", err
+ }
+ if _, err := conn.Write(data); err != nil {
+ return "", fmt.Errorf("send route request failed: %w", err)
+ }
+
+ if err := conn.SetReadDeadline(time.Now().Add(timeout)); err != nil {
+ return "", err
+ }
+ frameLenBuf := make([]byte, 4)
+ if _, err := io.ReadFull(conn, frameLenBuf); err != nil {
+ return "", fmt.Errorf("read frame length failed: %w", err)
+ }
+ frameLen := int(binary.BigEndian.Uint32(frameLenBuf))
+ if frameLen < 8 {
+ return "", fmt.Errorf("name server %s response frame too short:
%d", nameServerAddr, frameLen)
+ }
+ if frameLen > maxFrameSize {
+ return "", fmt.Errorf("name server %s response frame too large:
%d", nameServerAddr, frameLen)
+ }
+
+ frameBuf := make([]byte, frameLen)
+ if _, err := io.ReadFull(conn, frameBuf); err != nil {
+ return "", fmt.Errorf("read frame data failed: %w", err)
+ }
+
+ if len(frameBuf) < 4 {
+ return "", fmt.Errorf("response frame too short")
+ }
+ oriHeaderLen := binary.BigEndian.Uint32(frameBuf[0:4])
+ headerLen := int(oriHeaderLen & 0xFFFFFF)
+ if len(frameBuf) < 4+headerLen {
+ return "", fmt.Errorf("response header truncated")
+ }
+
+ respCode := int16(binary.BigEndian.Uint16(frameBuf[4:6]))
+ if respCode != 0 {
+ return "", fmt.Errorf("name server %s returned error code %d
for route query", nameServerAddr, respCode)
+ }
+
+ _, err = decodeResponseHeader(frameBuf[4 : 4+headerLen])
+ if err != nil {
+ return "", fmt.Errorf("decode response header failed: %w", err)
+ }
+
+ bodyStart := 4 + headerLen
+ if bodyStart >= len(frameBuf) {
+ return "", fmt.Errorf("response has no body")
+ }
+ body := frameBuf[bodyStart:]
+
+ return parseBrokerAddrFromRouteBody(body, brokerName)
+}
+
+func decodeResponseHeader(data []byte) (map[string]string, error) {
+ buf := bytes.NewReader(data)
+
+ if buf.Len() < rmqResponseHeaderFixedPartLength {
+ return nil, fmt.Errorf("header too short for fixed fields")
+ }
+ discard := make([]byte, rmqResponseHeaderFixedPartLength)
+ if _, err := io.ReadFull(buf, discard); err != nil {
+ return nil, err
+ }
+
+ var remarkLen int32
+ if err := binary.Read(buf, binary.BigEndian, &remarkLen); err != nil {
+ return nil, err
+ }
+ if remarkLen < 0 || int(remarkLen) > buf.Len() {
+ return nil, fmt.Errorf("invalid remark length: %d", remarkLen)
+ }
+ if remarkLen > 0 {
+ discardRemark := make([]byte, remarkLen)
+ if _, err := io.ReadFull(buf, discardRemark); err != nil {
+ return nil, err
+ }
+ }
+
+ var extLen int32
+ if err := binary.Read(buf, binary.BigEndian, &extLen); err != nil {
+ return nil, err
+ }
+ if extLen < 0 || int(extLen) > buf.Len() {
+ return nil, fmt.Errorf("invalid ext fields length: %d", extLen)
+ }
+ extFields := make(map[string]string)
+ if extLen > 0 {
+ extData := make([]byte, extLen)
+ if _, err := io.ReadFull(buf, extData); err != nil {
+ return nil, err
+ }
+ extBuf := bytes.NewReader(extData)
+ for extBuf.Len() > 0 {
+ var kLen int16
+ if err := binary.Read(extBuf, binary.BigEndian, &kLen);
err != nil {
+ return nil, fmt.Errorf("read ext key length
failed: %w", err)
+ }
+ if kLen < 0 || int(kLen) > extBuf.Len() {
+ return nil, fmt.Errorf("invalid ext key length:
%d", kLen)
+ }
+ key := make([]byte, kLen)
+ if _, err := io.ReadFull(extBuf, key); err != nil {
+ return nil, fmt.Errorf("read ext key failed:
%w", err)
+ }
+ var vLen int32
+ if err := binary.Read(extBuf, binary.BigEndian, &vLen);
err != nil {
+ return nil, fmt.Errorf("read ext value length
failed: %w", err)
+ }
+ if vLen < 0 || int(vLen) > extBuf.Len() {
+ return nil, fmt.Errorf("invalid ext value
length: %d", vLen)
+ }
+ value := make([]byte, vLen)
+ if _, err := io.ReadFull(extBuf, value); err != nil {
+ return nil, fmt.Errorf("read ext value failed:
%w", err)
+ }
+ extFields[string(key)] = string(value)
+ }
+ }
+
+ return extFields, nil
+}
+
+type topicRouteData struct {
+ QueueDataList []queueData `json:"queueDatas"`
+ BrokerDataList []brokerData `json:"brokerDatas"`
+}
+
+type queueData struct {
+ BrokerName string `json:"brokerName"`
+}
+
+type brokerData struct {
+ BrokerName string `json:"brokerName"`
+ BrokerAddrs map[string]string `json:"brokerAddrs"`
+}
+
+func parseBrokerAddrFromRouteBody(body []byte, brokerName string) (string,
error) {
+ var route topicRouteData
+ if err := json.Unmarshal(body, &route); err != nil {
+ return "", fmt.Errorf("unmarshal topic route data failed: %w",
err)
+ }
+
+ for _, bd := range route.BrokerDataList {
+ if bd.BrokerName == brokerName {
+ if addr, ok := bd.BrokerAddrs["0"]; ok {
+ return addr, nil
+ }
+ return "", fmt.Errorf("broker %s master (addr key '0')
not found in route data", brokerName)
+ }
+ }
+
+ return "", fmt.Errorf("broker %s not found in route data", brokerName)
+}
+
+func getQueueOffsetFromActionContext(actionCtx map[string]interface{}) int64 {
+ v, ok := actionCtx[ActionContextKeyQueueOffset]
+ if !ok {
+ return 0
+ }
+ switch val := v.(type) {
+ case float64:
+ return int64(val)
+ case int64:
+ return val
+ case int:
+ return int64(val)
+ case json.Number:
+ n, _ := val.Int64()
+ return n
+ default:
+ return 0
+ }
+}
diff --git a/pkg/integration/rocketmq/end_transaction_sender_test.go
b/pkg/integration/rocketmq/end_transaction_sender_test.go
new file mode 100644
index 00000000..4b7050f9
--- /dev/null
+++ b/pkg/integration/rocketmq/end_transaction_sender_test.go
@@ -0,0 +1,652 @@
+/*
+ * 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.
+ */
+
+package rocketmq
+
+import (
+ "bytes"
+ "encoding/binary"
+ "encoding/json"
+ "errors"
+ "io"
+ "net"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+type stubBrokerAddrResolver struct {
+ addr string
+ err error
+
+ calls int
+ lastNsAddrs []string
+ lastTopic string
+ lastBroker string
+ lastTimeout time.Duration
+}
+
+func (s *stubBrokerAddrResolver) ResolveBrokerAddr(nameServerAddrs []string,
topic string, brokerName string, timeout time.Duration) (string, error) {
+ s.calls++
+ s.lastNsAddrs = nameServerAddrs
+ s.lastTopic = topic
+ s.lastBroker = brokerName
+ s.lastTimeout = timeout
+ if s.err != nil {
+ return "", s.err
+ }
+ return s.addr, nil
+}
+
+type stubTCPSender struct {
+ err error
+
+ calls int
+ lastAddr string
+ lastData []byte
+}
+
+func (s *stubTCPSender) Send(addr string, data []byte, timeout time.Duration)
error {
+ s.calls++
+ s.lastAddr = addr
+ s.lastData = data
+ if s.err != nil {
+ return s.err
+ }
+ return nil
+}
+
+func (s *stubTCPSender) Close() error {
+ return nil
+}
+
+func TestEndTransactionRequestHeader_Encode(t *testing.T) {
+ header := &endTransactionRequestHeader{
+ Topic: "test-topic",
+ ProducerGroup: "test-group",
+ TranStateTableOffset: 42,
+ CommitLogOffset: 1024,
+ CommitOrRollback: commitOrRollbackCommit,
+ FromTransactionCheck: false,
+ MsgID: "msg-001",
+ TransactionId: "tx-001",
+ }
+
+ result := header.Encode()
+
+ assert.Equal(t, "test-topic", result["topic"])
+ assert.Equal(t, "test-group", result["producerGroup"])
+ assert.Equal(t, "42", result["tranStateTableOffset"])
+ assert.Equal(t, "1024", result["commitLogOffset"])
+ assert.Equal(t, "8", result["commitOrRollback"])
+ assert.Equal(t, "false", result["fromTransactionCheck"])
+ assert.Equal(t, "msg-001", result["msgId"])
+ assert.Equal(t, "tx-001", result["transactionId"])
+}
+
+func TestEndTransactionRequestHeader_EncodeRollback(t *testing.T) {
+ header := &endTransactionRequestHeader{
+ Topic: "test-topic",
+ ProducerGroup: "test-group",
+ TranStateTableOffset: 10,
+ CommitLogOffset: 2048,
+ CommitOrRollback: commitOrRollbackRollback,
+ FromTransactionCheck: false,
+ MsgID: "msg-002",
+ TransactionId: "tx-002",
+ }
+
+ result := header.Encode()
+
+ assert.Equal(t, "12", result["commitOrRollback"])
+}
+
+func TestNewEndTransactionCommand(t *testing.T) {
+ header := &endTransactionRequestHeader{
+ Topic: "test-topic",
+ ProducerGroup: "test-group",
+ CommitOrRollback: commitOrRollbackCommit,
+ MsgID: "msg-001",
+ TransactionId: "tx-001",
+ }
+
+ cmd := newEndTransactionCommand(header)
+
+ assert.Equal(t, reqEndTransaction, cmd.Code)
+ assert.Equal(t, rmqLanguageGo, cmd.Language)
+ assert.Equal(t, rmqProtocolVersion, cmd.Version)
+ assert.NotZero(t, cmd.Opaque)
+ assert.Equal(t, "test-topic", cmd.ExtFields["topic"])
+ assert.Equal(t, "test-group", cmd.ExtFields["producerGroup"])
+ assert.Equal(t, "8", cmd.ExtFields["commitOrRollback"])
+}
+
+func TestRemotingCommand_Encode_FrameFormat(t *testing.T) {
+ cmd := &remotingCommand{
+ Code: reqEndTransaction,
+ Language: rmqLanguageGo,
+ Version: rmqProtocolVersion,
+ Opaque: 1,
+ Flag: 0,
+ Remark: "",
+ ExtFields: map[string]string{},
+ }
+
+ data, err := cmd.encode()
+ require.NoError(t, err)
+ require.NotNil(t, data)
+
+ reader := bytes.NewReader(data)
+
+ var frameSize int32
+ err = binary.Read(reader, binary.BigEndian, &frameSize)
+ require.NoError(t, err)
+ assert.Equal(t, int32(len(data)-4), frameSize)
+
+ var headerLenRaw int32
+ err = binary.Read(reader, binary.BigEndian, &headerLenRaw)
+ require.NoError(t, err)
+ codecType := byte((headerLenRaw >> 24) & 0xFF)
+ headerLen := headerLenRaw & 0xFFFFFF
+ assert.Equal(t, rmqCodecType, codecType)
+ assert.Equal(t, int32(rmqHeaderFixedLength), headerLen)
+
+ var code int16
+ err = binary.Read(reader, binary.BigEndian, &code)
+ require.NoError(t, err)
+ assert.Equal(t, reqEndTransaction, code)
+
+ language, err := reader.ReadByte()
+ require.NoError(t, err)
+ assert.Equal(t, rmqLanguageGo, language)
+
+ var version int16
+ err = binary.Read(reader, binary.BigEndian, &version)
+ require.NoError(t, err)
+ assert.Equal(t, rmqProtocolVersion, version)
+
+ var opaque int32
+ err = binary.Read(reader, binary.BigEndian, &opaque)
+ require.NoError(t, err)
+ assert.Equal(t, int32(1), opaque)
+
+ var flag int32
+ err = binary.Read(reader, binary.BigEndian, &flag)
+ require.NoError(t, err)
+ assert.Equal(t, int32(0), flag)
+
+ var remarkLen int32
+ err = binary.Read(reader, binary.BigEndian, &remarkLen)
+ require.NoError(t, err)
+ assert.Equal(t, int32(0), remarkLen)
+
+ var extLen int32
+ err = binary.Read(reader, binary.BigEndian, &extLen)
+ require.NoError(t, err)
+ assert.Equal(t, int32(0), extLen)
+}
+
+func TestRemotingCommand_Encode_WithExtFields(t *testing.T) {
+ cmd := &remotingCommand{
+ Code: reqEndTransaction,
+ Language: rmqLanguageGo,
+ Version: rmqProtocolVersion,
+ Opaque: 42,
+ ExtFields: map[string]string{
+ "producerGroup": "test-group",
+ },
+ }
+
+ data, err := cmd.encode()
+ require.NoError(t, err)
+ require.NotNil(t, data)
+
+ frameSize := int32(binary.BigEndian.Uint32(data[0:4]))
+ assert.Equal(t, int32(len(data)-4), frameSize)
+
+ headerLenRaw := binary.BigEndian.Uint32(data[4:8])
+ headerLen := int(headerLenRaw & 0xFFFFFF)
+ assert.Equal(t, rmqHeaderFixedLength+29, headerLen)
+}
+
+func TestEncodeExtFields_Empty(t *testing.T) {
+ result, err := encodeExtFields(map[string]string{})
+ require.NoError(t, err)
+ assert.Empty(t, result)
+}
+
+func TestEncodeExtFields_SingleField(t *testing.T) {
+ fields := map[string]string{
+ "key": "value",
+ }
+
+ result, err := encodeExtFields(fields)
+ require.NoError(t, err)
+
+ reader := bytes.NewReader(result)
+
+ var keyLen int16
+ err = binary.Read(reader, binary.BigEndian, &keyLen)
+ require.NoError(t, err)
+ assert.Equal(t, int16(3), keyLen)
+
+ keyBuf := make([]byte, keyLen)
+ _, err = reader.Read(keyBuf)
+ require.NoError(t, err)
+ assert.Equal(t, "key", string(keyBuf))
+
+ var valueLen int32
+ err = binary.Read(reader, binary.BigEndian, &valueLen)
+ require.NoError(t, err)
+ assert.Equal(t, int32(5), valueLen)
+
+ valueBuf := make([]byte, valueLen)
+ _, err = reader.Read(valueBuf)
+ require.NoError(t, err)
+ assert.Equal(t, "value", string(valueBuf))
+}
+
+func TestMarkProtocolType(t *testing.T) {
+ result := markProtocolType(100)
+
+ assert.Equal(t, rmqCodecType, result[0])
+ assert.Equal(t, byte(0x00), result[1])
+ assert.Equal(t, byte(0x00), result[2])
+ assert.Equal(t, byte(0x64), result[3])
+}
+
+func TestParseBrokerAddrFromRouteBody_FoundMaster(t *testing.T) {
+ route := topicRouteData{
+ BrokerDataList: []brokerData{
+ {
+ BrokerName: "broker-a",
+ BrokerAddrs: map[string]string{
+ "0": "192.168.1.100:10911",
+ "1": "192.168.1.101:10911",
+ },
+ },
+ },
+ }
+ body, _ := json.Marshal(route)
+
+ addr, err := parseBrokerAddrFromRouteBody(body, "broker-a")
+
+ require.NoError(t, err)
+ assert.Equal(t, "192.168.1.100:10911", addr)
+}
+
+func TestParseBrokerAddrFromRouteBody_MasterNotFound(t *testing.T) {
+ route := topicRouteData{
+ BrokerDataList: []brokerData{
+ {
+ BrokerName: "broker-a",
+ BrokerAddrs: map[string]string{
+ "1": "192.168.1.101:10911",
+ },
+ },
+ },
+ }
+ body, _ := json.Marshal(route)
+
+ _, err := parseBrokerAddrFromRouteBody(body, "broker-a")
+
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "master (addr key '0') not found")
+}
+
+func TestParseBrokerAddrFromRouteBody_BrokerNotFound(t *testing.T) {
+ route := topicRouteData{
+ BrokerDataList: []brokerData{
+ {
+ BrokerName: "broker-a",
+ BrokerAddrs: map[string]string{"0":
"192.168.1.100:10911"},
+ },
+ },
+ }
+ body, _ := json.Marshal(route)
+
+ _, err := parseBrokerAddrFromRouteBody(body, "broker-b")
+
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "broker-b")
+}
+
+func TestParseBrokerAddrFromRouteBody_InvalidJSON(t *testing.T) {
+ _, err := parseBrokerAddrFromRouteBody([]byte("invalid json"),
"broker-a")
+
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "unmarshal")
+}
+
+func TestGetQueueOffsetFromActionContext_Float64(t *testing.T) {
+ ctx := map[string]interface{}{
+ ActionContextKeyQueueOffset: float64(42),
+ }
+ assert.Equal(t, int64(42), getQueueOffsetFromActionContext(ctx))
+}
+
+func TestGetQueueOffsetFromActionContext_Int64(t *testing.T) {
+ ctx := map[string]interface{}{
+ ActionContextKeyQueueOffset: int64(42),
+ }
+ assert.Equal(t, int64(42), getQueueOffsetFromActionContext(ctx))
+}
+
+func TestGetQueueOffsetFromActionContext_Int(t *testing.T) {
+ ctx := map[string]interface{}{
+ ActionContextKeyQueueOffset: int(42),
+ }
+ assert.Equal(t, int64(42), getQueueOffsetFromActionContext(ctx))
+}
+
+func TestGetQueueOffsetFromActionContext_Missing(t *testing.T) {
+ ctx := map[string]interface{}{}
+ assert.Equal(t, int64(0), getQueueOffsetFromActionContext(ctx))
+}
+
+func TestGetQueueOffsetFromActionContext_InvalidType(t *testing.T) {
+ ctx := map[string]interface{}{
+ ActionContextKeyQueueOffset: "not a number",
+ }
+ assert.Equal(t, int64(0), getQueueOffsetFromActionContext(ctx))
+}
+
+func TestGetStringFromMap_Found(t *testing.T) {
+ ctx := map[string]interface{}{
+ ActionContextKeyMsgId: "msg-001",
+ }
+ assert.Equal(t, "msg-001", getStringFromMap(ctx, ActionContextKeyMsgId))
+}
+
+func TestGetStringFromMap_Missing(t *testing.T) {
+ ctx := map[string]interface{}{}
+ assert.Equal(t, "", getStringFromMap(ctx, ActionContextKeyMsgId))
+}
+
+func TestGetStringFromMap_WrongType(t *testing.T) {
+ ctx := map[string]interface{}{
+ ActionContextKeyMsgId: 12345,
+ }
+ assert.Equal(t, "", getStringFromMap(ctx, ActionContextKeyMsgId))
+}
+
+func TestSendEndTransaction_Success(t *testing.T) {
+ resolver := &stubBrokerAddrResolver{addr: "192.168.1.100:10911"}
+ sender := &stubTCPSender{}
+ header := &endTransactionRequestHeader{
+ Topic: "test-topic",
+ ProducerGroup: "test-group",
+ TranStateTableOffset: 42,
+ CommitLogOffset: 1024,
+ CommitOrRollback: commitOrRollbackCommit,
+ MsgID: "msg-001",
+ TransactionId: "tx-001",
+ }
+
+ err := sendEndTransaction(
+ []string{"nameserver:9876"},
+ "test-topic",
+ "broker-a",
+ header,
+ 3*time.Second,
+ resolver,
+ sender,
+ )
+
+ require.NoError(t, err)
+ assert.Equal(t, 1, resolver.calls)
+ assert.Equal(t, []string{"nameserver:9876"}, resolver.lastNsAddrs)
+ assert.Equal(t, "test-topic", resolver.lastTopic)
+ assert.Equal(t, "broker-a", resolver.lastBroker)
+ assert.Equal(t, 1, sender.calls)
+ assert.Equal(t, "192.168.1.100:10911", sender.lastAddr)
+ assert.NotEmpty(t, sender.lastData)
+
+ frameSize := int32(binary.BigEndian.Uint32(sender.lastData[0:4]))
+ assert.Equal(t, int32(len(sender.lastData)-4), frameSize)
+}
+
+func TestSendEndTransaction_ResolverError(t *testing.T) {
+ resolver := &stubBrokerAddrResolver{err: errors.New("name server
unreachable")}
+ sender := &stubTCPSender{}
+ header := &endTransactionRequestHeader{
+ Topic: "test-topic",
+ ProducerGroup: "test-group",
+ }
+
+ err := sendEndTransaction(
+ []string{"nameserver:9876"},
+ "test-topic",
+ "broker-a",
+ header,
+ 3*time.Second,
+ resolver,
+ sender,
+ )
+
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "resolve broker addr failed")
+ assert.Equal(t, 0, sender.calls)
+}
+
+func TestSendEndTransaction_SendError(t *testing.T) {
+ resolver := &stubBrokerAddrResolver{addr: "192.168.1.100:10911"}
+ sender := &stubTCPSender{err: errors.New("connection refused")}
+ header := &endTransactionRequestHeader{
+ Topic: "test-topic",
+ ProducerGroup: "test-group",
+ }
+
+ err := sendEndTransaction(
+ []string{"nameserver:9876"},
+ "test-topic",
+ "broker-a",
+ header,
+ 3*time.Second,
+ resolver,
+ sender,
+ )
+
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "send to broker")
+ assert.Equal(t, 1, resolver.calls)
+ assert.Equal(t, 1, sender.calls)
+}
+
+func TestDefaultBrokerAddrResolver_TriesAllNameServers(t *testing.T) {
+ resolver := &defaultBrokerAddrResolver{}
+
+ _, err := resolver.ResolveBrokerAddr(
+ []string{"127.0.0.1:1", "127.0.0.1:1"},
+ "test-topic",
+ "broker-a",
+ 3*time.Second,
+ )
+
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "broker")
+}
+
+func TestConnPool_GetAndPut(t *testing.T) {
+ // Start a temporary TCP listener to serve as a fake broker.
+ ln, err := net.Listen("tcp", "127.0.0.1:0")
+ require.NoError(t, err)
+ defer ln.Close()
+
+ go func() {
+ for {
+ conn, err := ln.Accept()
+ if err != nil {
+ return
+ }
+ go func(c net.Conn) {
+ defer c.Close()
+ buf := make([]byte, 4)
+ if _, err := io.ReadFull(c, buf); err != nil {
+ return
+ }
+ frameLen := binary.BigEndian.Uint32(buf)
+ payload := make([]byte, frameLen)
+ if _, err := io.ReadFull(c, payload); err !=
nil {
+ return
+ }
+ // Write back a success response (code=0).
+ resp := make([]byte, 8)
+ binary.BigEndian.PutUint32(resp[0:4], 4)
+ binary.BigEndian.PutUint16(resp[4:6], 0)
+ c.Write(resp)
+ }(conn)
+ }
+ }()
+
+ pool := newConnPool(ln.Addr().String(), 2)
+
+ // First get should create a new connection.
+ conn1, isNew1, err := pool.get(3 * time.Second)
+ require.NoError(t, err)
+ require.NotNil(t, conn1)
+ assert.True(t, isNew1)
+
+ // Return the connection to the pool.
+ pool.put(conn1)
+
+ // Second get should reuse the pooled connection.
+ conn2, isNew2, err := pool.get(3 * time.Second)
+ require.NoError(t, err)
+ require.NotNil(t, conn2)
+ assert.False(t, isNew2)
+ assert.Equal(t, conn1.RemoteAddr().String(),
conn2.RemoteAddr().String())
+
+ conn2.Close()
+}
+
+func TestConnPool_PutWhenFullClosesConnection(t *testing.T) {
+ ln, err := net.Listen("tcp", "127.0.0.1:0")
+ require.NoError(t, err)
+ defer ln.Close()
+
+ pool := newConnPool(ln.Addr().String(), 2)
+
+ conn, err := net.Dial("tcp", ln.Addr().String())
+ require.NoError(t, err)
+
+ // Fill the pool.
+ for i := 0; i < 2; i++ {
+ c, err := net.Dial("tcp", ln.Addr().String())
+ require.NoError(t, err)
+ pool.put(c)
+ }
+
+ // Putting one more should close it instead of overflowing.
+ pool.put(conn)
+ // The connection should eventually be closed; we just verify no panic.
+}
+
+func TestConnPool_PutAfterClose(t *testing.T) {
+ ln, err := net.Listen("tcp", "127.0.0.1:0")
+ require.NoError(t, err)
+ defer ln.Close()
+
+ pool := newConnPool(ln.Addr().String(), 2)
+
+ conn, err := net.Dial("tcp", ln.Addr().String())
+ require.NoError(t, err)
+
+ pool.closeAll()
+
+ // After closeAll, put should close the connection instead of returning
it to the pool.
+ pool.put(conn)
+ // No panic and connection is closed; verified implicitly.
+}
+
+func TestConnPool_GetDialsWhenEmpty(t *testing.T) {
+ ln, err := net.Listen("tcp", "127.0.0.1:0")
+ require.NoError(t, err)
+ defer ln.Close()
+
+ pool := newConnPool(ln.Addr().String(), 2)
+
+ // Get from an empty pool should dial a new connection.
+ conn, isNew, err := pool.get(3 * time.Second)
+ require.NoError(t, err)
+ require.NotNil(t, conn)
+ assert.True(t, isNew)
+ conn.Close()
+}
+
+func TestDefaultTCPSender_RetryOnPooledConnFailure(t *testing.T) {
+ ln, err := net.Listen("tcp", "127.0.0.1:0")
+ require.NoError(t, err)
+ defer ln.Close()
+
+ go func() {
+ for {
+ conn, err := ln.Accept()
+ if err != nil {
+ return
+ }
+ go func(c net.Conn) {
+ defer c.Close()
+ buf := make([]byte, 4)
+ if _, err := io.ReadFull(c, buf); err != nil {
+ return
+ }
+ frameLen := binary.BigEndian.Uint32(buf)
+ payload := make([]byte, frameLen)
+ if _, err := io.ReadFull(c, payload); err !=
nil {
+ return
+ }
+ resp, _ := (&remotingCommand{
+ Code: 0,
+ Language: rmqLanguageGo,
+ Version: rmqProtocolVersion,
+ Opaque: 1,
+ Flag: 0,
+ Remark: "",
+ ExtFields: map[string]string{},
+ }).encode()
+ c.Write(resp)
+ }(conn)
+ }
+ }()
+
+ sender := newDefaultTCPSender(2)
+
+ // Dial a connection, immediately close it, and put the closed conn
into the pool.
+ staleConn, err := net.Dial("tcp", ln.Addr().String())
+ require.NoError(t, err)
+ staleConn.Close()
+
+ poolAny, _ := sender.pools.LoadOrStore(ln.Addr().String(),
newConnPool(ln.Addr().String(), 2))
+ pool := poolAny.(*connPool)
+ pool.put(staleConn)
+
+ data, err := (&remotingCommand{
+ Code: reqEndTransaction,
+ Language: rmqLanguageGo,
+ Version: rmqProtocolVersion,
+ Opaque: 1,
+ ExtFields: map[string]string{},
+ }).encode()
+ require.NoError(t, err)
+
+ // Send should fail on the stale pooled conn, then retry with a fresh
conn and succeed.
+ err = sender.Send(ln.Addr().String(), data, 3*time.Second)
+ require.NoError(t, err)
+}
diff --git a/pkg/integration/rocketmq/seata_producer.go
b/pkg/integration/rocketmq/seata_producer.go
index 3a22f8e6..5de336fe 100644
--- a/pkg/integration/rocketmq/seata_producer.go
+++ b/pkg/integration/rocketmq/seata_producer.go
@@ -21,6 +21,7 @@ import (
"context"
"fmt"
"sync"
+ "time"
"github.com/apache/rocketmq-client-go/v2/primitive"
"github.com/apache/rocketmq-client-go/v2/producer"
@@ -70,6 +71,13 @@ func NewSeataMQProducer(cfg *SeataMQProducerConfig)
(*SeataMQProducer, error) {
return nil, fmt.Errorf("GroupName cannot be empty")
}
+ if cfg.SendMsgTimeout <= 0 {
+ cfg.SendMsgTimeout = 3 * time.Second
+ }
+ if cfg.ConnPoolSize <= 0 {
+ cfg.ConnPoolSize = 4
+ }
+
p := &SeataMQProducer{
config: cfg,
}
@@ -139,6 +147,11 @@ func (p *SeataMQProducer) Shutdown() error {
if err := p.normalProducer.Shutdown(); err != nil {
errs = append(errs, err)
}
+ if p.tccAction != nil && p.tccAction.sender != nil {
+ if err := p.tccAction.sender.Close(); err != nil {
+ errs = append(errs, err)
+ }
+ }
if len(errs) > 0 {
return fmt.Errorf("shutdown errors: %v", errs)
diff --git a/pkg/integration/rocketmq/tcc_rocketmq_action.go
b/pkg/integration/rocketmq/tcc_rocketmq_action.go
index a769646d..8ff7afc8 100644
--- a/pkg/integration/rocketmq/tcc_rocketmq_action.go
+++ b/pkg/integration/rocketmq/tcc_rocketmq_action.go
@@ -30,11 +30,19 @@ import (
type TCCRocketMQAction struct {
producer *SeataMQProducer
+ resolver brokerAddrResolver
+ sender tcpSender
}
func NewTCCRocketMQAction(producer *SeataMQProducer) *TCCRocketMQAction {
+ poolSize := defaultConnPoolSize
+ if producer != nil && producer.config != nil &&
producer.config.ConnPoolSize > 0 {
+ poolSize = producer.config.ConnPoolSize
+ }
return &TCCRocketMQAction{
producer: producer,
+ resolver: &defaultBrokerAddrResolver{},
+ sender: newDefaultTCPSender(poolSize),
}
}
@@ -78,6 +86,7 @@ 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)
@@ -85,20 +94,110 @@ func (a *TCCRocketMQAction) Prepare(ctx context.Context,
params interface{}) (bo
}
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, ok := a.buildEndTransactionHeader(bac, topic,
commitOrRollbackCommit)
+ if !ok {
+ log.Warnf("[TCCRocketMQ] Commit cannot resolve valid
commitLogOffset, skip active END_TRANSACTION, fallback to check-back, xid=%s,
branchId=%d", bac.Xid, bac.BranchId)
+ return true, nil
+ }
+ 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, ok := a.buildEndTransactionHeader(bac, topic,
commitOrRollbackRollback)
+ if !ok {
+ log.Warnf("[TCCRocketMQ] Rollback cannot resolve valid
commitLogOffset, skip active END_TRANSACTION, fallback to check-back, xid=%s,
branchId=%d", bac.Xid, bac.BranchId)
+ return true, nil
+ }
+ 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, bool) {
+ actionCtx := bac.ActionContext
+ if actionCtx == nil {
+ actionCtx = make(map[string]interface{})
+ }
+
+ offsetMsgID := getStringFromMap(actionCtx, ActionContextKeyOffsetMsgId)
+ commitLogOffset := int64(0)
+ if offsetMsgID != "" {
+ msgID, err := primitive.UnmarshalMsgID([]byte(offsetMsgID))
+ if err == nil {
+ commitLogOffset = msgID.Offset
+ }
+ }
+
+ if commitLogOffset == 0 {
+ if msgIDStr := getStringFromMap(actionCtx,
ActionContextKeyMsgId); msgIDStr != "" {
+ if msgID, err :=
primitive.UnmarshalMsgID([]byte(msgIDStr)); err == nil {
+ commitLogOffset = msgID.Offset
+ }
+ }
+ }
+
+ if commitLogOffset == 0 {
+ return nil, false
+ }
+
+ return &endTransactionRequestHeader{
+ Topic: topic,
+ ProducerGroup: a.producer.config.GroupName,
+ TranStateTableOffset:
getQueueOffsetFromActionContext(actionCtx),
+ CommitLogOffset: commitLogOffset,
+ CommitOrRollback: commitOrRollback,
+ FromTransactionCheck: false,
+ MsgID: getStringFromMap(actionCtx,
ActionContextKeyMsgId),
+ TransactionId: getStringFromMap(actionCtx,
ActionContextKeyTransactionId),
+ }, true
+}
diff --git a/pkg/integration/rocketmq/tcc_rocketmq_action_test.go
b/pkg/integration/rocketmq/tcc_rocketmq_action_test.go
index b3e24267..651f58e3 100644
--- a/pkg/integration/rocketmq/tcc_rocketmq_action_test.go
+++ b/pkg/integration/rocketmq/tcc_rocketmq_action_test.go
@@ -19,7 +19,9 @@ package rocketmq
import (
"context"
+ "errors"
"testing"
+ "time"
"github.com/apache/rocketmq-client-go/v2/primitive"
"github.com/stretchr/testify/assert"
@@ -88,6 +90,315 @@ func
TestTCCRocketMQActionPrepare_SetsMessagePropertiesAndActionContext(t *testi
assert.Equal(t, "tx-1",
bac.ActionContext[ActionContextKeyTransactionId])
assert.Equal(t, 3, bac.ActionContext[ActionContextKeyQueueId])
assert.Equal(t, "broker-a",
bac.ActionContext[ActionContextKeyBrokerName])
+ assert.Equal(t, "topic-test", bac.ActionContext[ActionContextKeyTopic])
assert.Equal(t, 1, transactionProducer.sendCalls)
assert.Same(t, msg, transactionProducer.lastMsg)
}
+
+func TestTCCRocketMQAction_Commit_Success(t *testing.T) {
+ resolver := &stubBrokerAddrResolver{addr: "192.168.1.100:10911"}
+ sender := &stubTCPSender{}
+ action := &TCCRocketMQAction{
+ producer: &SeataMQProducer{
+ config: &SeataMQProducerConfig{
+ NameServerAddrs: []string{"nameserver:9876"},
+ GroupName: "test-group",
+ SendMsgTimeout: 3 * time.Second,
+ },
+ },
+ resolver: resolver,
+ sender: sender,
+ }
+ validMsgID := primitive.CreateMessageId([]byte{10, 93, 233, 58}, 10911,
42)
+ bac := &tm.BusinessActionContext{
+ Xid: "xid-123",
+ BranchId: 1001,
+ ActionContext: map[string]interface{}{
+ ActionContextKeyMsgId: validMsgID,
+ ActionContextKeyOffsetMsgId: "offset-1",
+ ActionContextKeyQueueOffset: int64(11),
+ ActionContextKeyTransactionId: "tx-1",
+ ActionContextKeyQueueId: 3,
+ ActionContextKeyBrokerName: "broker-a",
+ ActionContextKeyTopic: "topic-test",
+ },
+ }
+
+ ok, err := action.Commit(context.Background(), bac)
+
+ require.NoError(t, err)
+ assert.True(t, ok)
+ assert.Equal(t, 1, resolver.calls)
+ assert.Equal(t, "topic-test", resolver.lastTopic)
+ assert.Equal(t, "broker-a", resolver.lastBroker)
+ assert.Equal(t, 1, sender.calls)
+ assert.Equal(t, "192.168.1.100:10911", sender.lastAddr)
+}
+
+func TestTCCRocketMQAction_Commit_ErrorOnResolverError(t *testing.T) {
+ resolver := &stubBrokerAddrResolver{err: errors.New("name server
unreachable")}
+ sender := &stubTCPSender{}
+ action := &TCCRocketMQAction{
+ producer: &SeataMQProducer{
+ config: &SeataMQProducerConfig{
+ NameServerAddrs: []string{"nameserver:9876"},
+ GroupName: "test-group",
+ SendMsgTimeout: 3 * time.Second,
+ },
+ },
+ resolver: resolver,
+ sender: sender,
+ }
+ bac := &tm.BusinessActionContext{
+ Xid: "xid-123",
+ BranchId: 1001,
+ ActionContext: map[string]interface{}{
+ ActionContextKeyMsgId: "msg-1",
+ ActionContextKeyQueueOffset: int64(11),
+ ActionContextKeyTransactionId: "tx-1",
+ ActionContextKeyBrokerName: "broker-a",
+ ActionContextKeyTopic: "topic-test",
+ },
+ }
+
+ ok, err := action.Commit(context.Background(), bac)
+
+ require.NoError(t, err)
+ assert.True(t, ok)
+ assert.Equal(t, 0, sender.calls)
+}
+
+func TestTCCRocketMQAction_Rollback_Success(t *testing.T) {
+ resolver := &stubBrokerAddrResolver{addr: "192.168.1.100:10911"}
+ sender := &stubTCPSender{}
+ action := &TCCRocketMQAction{
+ producer: &SeataMQProducer{
+ config: &SeataMQProducerConfig{
+ NameServerAddrs: []string{"nameserver:9876"},
+ GroupName: "test-group",
+ SendMsgTimeout: 3 * time.Second,
+ },
+ },
+ resolver: resolver,
+ sender: sender,
+ }
+ validMsgID := primitive.CreateMessageId([]byte{10, 93, 233, 58}, 10911,
42)
+ bac := &tm.BusinessActionContext{
+ Xid: "xid-456",
+ BranchId: 2002,
+ ActionContext: map[string]interface{}{
+ ActionContextKeyMsgId: validMsgID,
+ ActionContextKeyOffsetMsgId: "offset-2",
+ ActionContextKeyQueueOffset: int64(22),
+ ActionContextKeyTransactionId: "tx-2",
+ ActionContextKeyQueueId: 1,
+ ActionContextKeyBrokerName: "broker-b",
+ ActionContextKeyTopic: "topic-rollback",
+ },
+ }
+
+ ok, err := action.Rollback(context.Background(), bac)
+
+ require.NoError(t, err)
+ assert.True(t, ok)
+ assert.Equal(t, 1, resolver.calls)
+ assert.Equal(t, "topic-rollback", resolver.lastTopic)
+ assert.Equal(t, "broker-b", resolver.lastBroker)
+ assert.Equal(t, 1, sender.calls)
+ assert.Equal(t, "192.168.1.100:10911", sender.lastAddr)
+}
+
+func TestBuildEndTransactionHeader_ParseCommitLogOffset(t *testing.T) {
+ offsetMsgID := primitive.CreateMessageId([]byte{10, 93, 233, 58},
10911, 42)
+ action := &TCCRocketMQAction{
+ producer: &SeataMQProducer{
+ config: &SeataMQProducerConfig{
+ GroupName: "test-group",
+ },
+ },
+ }
+ bac := &tm.BusinessActionContext{
+ ActionContext: map[string]interface{}{
+ ActionContextKeyOffsetMsgId: offsetMsgID,
+ ActionContextKeyQueueOffset: int64(11),
+ ActionContextKeyMsgId: "msg-1",
+ ActionContextKeyTransactionId: "tx-1",
+ },
+ }
+
+ header, ok := action.buildEndTransactionHeader(bac, "test-topic",
commitOrRollbackCommit)
+
+ require.True(t, ok)
+ assert.Equal(t, "test-topic", header.Topic)
+ assert.Equal(t, "test-group", header.ProducerGroup)
+ assert.Equal(t, int64(11), header.TranStateTableOffset)
+ assert.Equal(t, int64(42), header.CommitLogOffset)
+ assert.Equal(t, commitOrRollbackCommit, header.CommitOrRollback)
+ assert.Equal(t, "msg-1", header.MsgID)
+ assert.Equal(t, "tx-1", header.TransactionId)
+ assert.False(t, header.FromTransactionCheck)
+}
+
+func TestBuildEndTransactionHeader_InvalidOffsetMsgID(t *testing.T) {
+ action := &TCCRocketMQAction{
+ producer: &SeataMQProducer{
+ config: &SeataMQProducerConfig{
+ GroupName: "test-group",
+ },
+ },
+ }
+ validMsgID := primitive.CreateMessageId([]byte{10, 93, 233, 58}, 10911,
42)
+ bac := &tm.BusinessActionContext{
+ ActionContext: map[string]interface{}{
+ ActionContextKeyOffsetMsgId: "invalid-id",
+ ActionContextKeyMsgId: validMsgID,
+ ActionContextKeyQueueOffset: int64(11),
+ },
+ }
+
+ header, ok := action.buildEndTransactionHeader(bac, "test-topic",
commitOrRollbackRollback)
+
+ require.True(t, ok)
+ assert.Equal(t, "test-topic", header.Topic)
+ assert.Equal(t, int64(42), header.CommitLogOffset)
+ assert.Equal(t, commitOrRollbackRollback, header.CommitOrRollback)
+}
+
+func TestBuildEndTransactionHeader_MissingOffsetMsgID(t *testing.T) {
+ action := &TCCRocketMQAction{
+ producer: &SeataMQProducer{
+ config: &SeataMQProducerConfig{
+ GroupName: "test-group",
+ },
+ },
+ }
+ bac := &tm.BusinessActionContext{
+ ActionContext: map[string]interface{}{
+ ActionContextKeyQueueOffset: int64(11),
+ },
+ }
+
+ header, ok := action.buildEndTransactionHeader(bac, "test-topic",
commitOrRollbackCommit)
+
+ require.False(t, ok)
+ require.Nil(t, header)
+}
+
+func TestTCCRocketMQAction_Rollback_ErrorOnSendError(t *testing.T) {
+ resolver := &stubBrokerAddrResolver{addr: "192.168.1.100:10911"}
+ sender := &stubTCPSender{err: errors.New("connection refused")}
+ action := &TCCRocketMQAction{
+ producer: &SeataMQProducer{
+ config: &SeataMQProducerConfig{
+ NameServerAddrs: []string{"nameserver:9876"},
+ GroupName: "test-group",
+ SendMsgTimeout: 3 * time.Second,
+ },
+ },
+ resolver: resolver,
+ sender: sender,
+ }
+ validMsgID := primitive.CreateMessageId([]byte{10, 93, 233, 58}, 10911,
42)
+ bac := &tm.BusinessActionContext{
+ Xid: "xid-789",
+ BranchId: 3003,
+ ActionContext: map[string]interface{}{
+ ActionContextKeyMsgId: validMsgID,
+ ActionContextKeyQueueOffset: int64(33),
+ ActionContextKeyTransactionId: "tx-3",
+ ActionContextKeyBrokerName: "broker-c",
+ ActionContextKeyTopic: "topic-send-err",
+ },
+ }
+
+ ok, err := action.Rollback(context.Background(), bac)
+
+ require.NoError(t, err)
+ assert.True(t, ok)
+ assert.Equal(t, 1, resolver.calls)
+ assert.Equal(t, 1, sender.calls)
+}
+
+func TestBuildEndTransactionHeader_MissingAllIds(t *testing.T) {
+ action := &TCCRocketMQAction{
+ producer: &SeataMQProducer{
+ config: &SeataMQProducerConfig{
+ GroupName: "test-group",
+ },
+ },
+ }
+ bac := &tm.BusinessActionContext{
+ ActionContext: map[string]interface{}{
+ ActionContextKeyQueueOffset: int64(11),
+ ActionContextKeyBrokerName: "broker-a",
+ ActionContextKeyTopic: "test-topic",
+ },
+ }
+
+ header, ok := action.buildEndTransactionHeader(bac, "test-topic",
commitOrRollbackCommit)
+
+ require.False(t, ok)
+ require.Nil(t, header)
+}
+
+func TestTCCRocketMQAction_Commit_InvalidCommitLogOffset(t *testing.T) {
+ resolver := &stubBrokerAddrResolver{addr: "192.168.1.100:10911"}
+ sender := &stubTCPSender{}
+ action := &TCCRocketMQAction{
+ producer: &SeataMQProducer{
+ config: &SeataMQProducerConfig{
+ NameServerAddrs: []string{"nameserver:9876"},
+ GroupName: "test-group",
+ SendMsgTimeout: 3 * time.Second,
+ },
+ },
+ resolver: resolver,
+ sender: sender,
+ }
+ bac := &tm.BusinessActionContext{
+ Xid: "xid-000",
+ BranchId: 9999,
+ ActionContext: map[string]interface{}{
+ ActionContextKeyBrokerName: "broker-a",
+ ActionContextKeyTopic: "topic-test",
+ // missing both offsetMsgId and msgId =>
commitLogOffset == 0
+ },
+ }
+
+ ok, err := action.Commit(context.Background(), bac)
+
+ require.NoError(t, err)
+ assert.True(t, ok)
+ assert.Equal(t, 0, sender.calls) // must skip sending END_TRANSACTION
+}
+
+func TestTCCRocketMQAction_Rollback_InvalidCommitLogOffset(t *testing.T) {
+ resolver := &stubBrokerAddrResolver{addr: "192.168.1.100:10911"}
+ sender := &stubTCPSender{}
+ action := &TCCRocketMQAction{
+ producer: &SeataMQProducer{
+ config: &SeataMQProducerConfig{
+ NameServerAddrs: []string{"nameserver:9876"},
+ GroupName: "test-group",
+ SendMsgTimeout: 3 * time.Second,
+ },
+ },
+ resolver: resolver,
+ sender: sender,
+ }
+ bac := &tm.BusinessActionContext{
+ Xid: "xid-111",
+ BranchId: 8888,
+ ActionContext: map[string]interface{}{
+ ActionContextKeyBrokerName: "broker-b",
+ ActionContextKeyTopic: "topic-rollback",
+ // missing both offsetMsgId and msgId =>
commitLogOffset == 0
+ },
+ }
+
+ ok, err := action.Rollback(context.Background(), bac)
+
+ require.NoError(t, err)
+ assert.True(t, ok)
+ assert.Equal(t, 0, sender.calls) // must skip sending END_TRANSACTION
+}
diff --git a/pkg/integration/rocketmq/transaction_listener.go
b/pkg/integration/rocketmq/transaction_listener.go
index 036b1482..f51cad1f 100644
--- a/pkg/integration/rocketmq/transaction_listener.go
+++ b/pkg/integration/rocketmq/transaction_listener.go
@@ -104,11 +104,11 @@ func mapGlobalStatusToLocalTransactionState(globalStatus
message.GlobalStatus) p
case message.GlobalStatusCommitted, message.GlobalStatusAsyncCommitting:
return primitive.CommitMessageState
case message.GlobalStatusRollbacked,
message.GlobalStatusTimeoutRollbacked, message.GlobalStatusRollbackFailed,
- message.GlobalStatusTimeoutRollbackFailed,
message.GlobalStatusCommitFailed:
+ message.GlobalStatusTimeoutRollbackFailed:
return primitive.RollbackMessageState
case message.GlobalStatusBegin, message.GlobalStatusCommitting,
message.GlobalStatusCommitRetrying,
message.GlobalStatusRollbacking,
message.GlobalStatusRollbackRetrying, message.GlobalStatusTimeoutRollbacking,
- message.GlobalStatusTimeoutRollbackRetrying,
message.GlobalStatusFinished:
+ message.GlobalStatusTimeoutRollbackRetrying,
message.GlobalStatusFinished, message.GlobalStatusCommitFailed:
return primitive.UnknowState
default:
return primitive.UnknowState
diff --git a/pkg/integration/rocketmq/transaction_listener_test.go
b/pkg/integration/rocketmq/transaction_listener_test.go
index 9fe6248d..a70aa3cd 100644
--- a/pkg/integration/rocketmq/transaction_listener_test.go
+++ b/pkg/integration/rocketmq/transaction_listener_test.go
@@ -73,7 +73,7 @@ func
TestSeataTransactionListenerCheckLocalTransaction_StatusMapping(t *testing.
{name: "commit retrying", globalStatus:
message.GlobalStatusCommitRetrying, expected: primitive.UnknowState},
{name: "rollbacking", globalStatus:
message.GlobalStatusRollbacking, expected: primitive.UnknowState},
{name: "timeout rollback retrying", globalStatus:
message.GlobalStatusTimeoutRollbackRetrying, expected: primitive.UnknowState},
- {name: "commit failed", globalStatus:
message.GlobalStatusCommitFailed, expected: primitive.RollbackMessageState},
+ {name: "commit failed", globalStatus:
message.GlobalStatusCommitFailed, expected: primitive.UnknowState},
{name: "timeout rollback failed", globalStatus:
message.GlobalStatusTimeoutRollbackFailed, expected:
primitive.RollbackMessageState},
{name: "finished", globalStatus: message.GlobalStatusFinished,
expected: primitive.UnknowState},
}
diff --git a/pkg/rm/tcc/tcc_service.go b/pkg/rm/tcc/tcc_service.go
index de4ef0b1..5c226689 100644
--- a/pkg/rm/tcc/tcc_service.go
+++ b/pkg/rm/tcc/tcc_service.go
@@ -21,6 +21,7 @@ import (
"context"
"encoding/json"
"errors"
+ "fmt"
"reflect"
"sync"
"time"
@@ -84,9 +85,52 @@ func (t *TCCServiceProxy) Prepare(ctx context.Context,
params interface{}) (inte
}
}
- // to set up the fence phase
tm.SetFencePhase(ctx, enum.FencePhasePrepare)
- return t.TCCResource.Prepare(ctx, params)
+ result, err := t.TCCResource.Prepare(ctx, params)
+ if err != nil {
+ return nil, err
+ }
+
+ bac := tm.GetBusinessActionContext(ctx)
+ if bac != nil && bac.IsDelayReport {
+ if err := t.reportActionContext(ctx, bac); err != nil {
+ log.Warnf("[TCC] report action context failed after
prepare, fallback to original branch applicationData (broker check-back may be
used), xid=%s, branchId=%d, err=%v", bac.Xid, bac.BranchId, err)
+ }
+ }
+
+ return result, nil
+}
+
+// reportActionContext reports the updated ActionContext to TC via
BranchReport.
+// It relies on the Seata Server (TC) behavior: AbstractCore.branchReport
updates
+// branchSession.applicationData, which will be delivered back to RM during
phase-two
+// (branchCommit/branchRollback). If this report fails, phase-two will fall
back to
+// broker check-back because the old applicationData lacks RocketMQ metadata.
+func (t *TCCServiceProxy) reportActionContext(ctx context.Context, bac
*tm.BusinessActionContext) error {
+ updatedActionContext := make(map[string]interface{})
+ for k, v := range bac.ActionContext {
+ updatedActionContext[k] = v
+ }
+ applicationData, err := json.Marshal(map[string]interface{}{
+ constant.ActionContext: updatedActionContext,
+ })
+ if err != nil {
+ 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{
+ BranchType: branch.BranchTypeTCC,
+ Xid: bac.Xid,
+ BranchId: bac.BranchId,
+ Status: branch.BranchStatusRegistered,
+ ApplicationData: string(applicationData),
+ })
+ if err != nil {
+ log.Errorf("[TCC] report updated ActionContext failed, xid=%s,
branchId=%d, err=%v", bac.Xid, bac.BranchId, err)
+ return fmt.Errorf("report updated ActionContext failed: %w",
err)
+ }
+ log.Infof("[TCC] report updated ActionContext success, xid=%s,
branchId=%d", bac.Xid, bac.BranchId)
+ return nil
}
// registeBranch send register branch transaction request
diff --git a/pkg/rm/tcc/tcc_service_test.go b/pkg/rm/tcc/tcc_service_test.go
index ddfd3f85..e5ec918c 100644
--- a/pkg/rm/tcc/tcc_service_test.go
+++ b/pkg/rm/tcc/tcc_service_test.go
@@ -19,6 +19,7 @@ package tcc
import (
"context"
+ "encoding/json"
"fmt"
"os"
"reflect"
@@ -30,8 +31,10 @@ import (
gostnet "github.com/dubbogo/gost/net"
"github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
"seata.apache.org/seata-go/v2/pkg/constant"
+ "seata.apache.org/seata-go/v2/pkg/protocol/branch"
"seata.apache.org/seata-go/v2/pkg/rm"
"seata.apache.org/seata-go/v2/pkg/rm/remoting/getty"
@@ -335,3 +338,55 @@ func TestTCCGetTransactionInfo(t1 *testing.T) {
func GetTestTwoPhaseService() rm.TwoPhaseInterface {
return &testdata2.TestTwoPhaseService{}
}
+
+func TestReportActionContext(t *testing.T) {
+ var capturedParam rm.BranchReportParam
+ patches :=
gomonkey.ApplyMethod(reflect.TypeOf(rm.GetRMRemotingInstance()),
"BranchReport", func(_ *getty.GettyRMRemoting, param rm.BranchReportParam)
error {
+ capturedParam = param
+ return nil
+ })
+ defer patches.Reset()
+
+ bac := &tm.BusinessActionContext{
+ Xid: "xid-report-1",
+ BranchId: 1001,
+ ActionContext: map[string]interface{}{
+ "msgId": "msg-001",
+ "topic": "test-topic",
+ },
+ }
+
+ err := testTccServiceProxy.reportActionContext(context.Background(),
bac)
+ require.NoError(t, err)
+ assert.Equal(t, "xid-report-1", capturedParam.Xid)
+ assert.Equal(t, int64(1001), capturedParam.BranchId)
+ assert.Equal(t, branch.BranchTypeTCC, capturedParam.BranchType)
+ assert.EqualValues(t, branch.BranchStatusRegistered,
capturedParam.Status)
+
+ var appData map[string]interface{}
+ err = json.Unmarshal([]byte(capturedParam.ApplicationData), &appData)
+ require.NoError(t, err)
+ actionCtx, ok := appData["actionContext"].(map[string]interface{})
+ require.True(t, ok)
+ assert.Equal(t, "msg-001", actionCtx["msgId"])
+ assert.Equal(t, "test-topic", actionCtx["topic"])
+}
+
+func TestReportActionContext_BranchReportFails(t *testing.T) {
+ patches :=
gomonkey.ApplyMethod(reflect.TypeOf(rm.GetRMRemotingInstance()),
"BranchReport", func(_ *getty.GettyRMRemoting, param rm.BranchReportParam)
error {
+ return fmt.Errorf("network error")
+ })
+ defer patches.Reset()
+
+ bac := &tm.BusinessActionContext{
+ Xid: "xid-report-2",
+ BranchId: 1002,
+ ActionContext: map[string]interface{}{
+ "key": "value",
+ },
+ }
+
+ err := testTccServiceProxy.reportActionContext(context.Background(),
bac)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "report updated ActionContext failed")
+}
diff --git a/pkg/util/flagext/day_test.go b/pkg/util/flagext/day_test.go
index d7a82bbd..0d8bd5e5 100644
--- a/pkg/util/flagext/day_test.go
+++ b/pkg/util/flagext/day_test.go
@@ -19,7 +19,6 @@ package flagext
import (
"testing"
- "time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -46,6 +45,7 @@ func TestDayValueYAML(t *testing.T) {
err = yaml.Unmarshal(expected, &actualStruct)
require.NoError(t, err)
assert.Equal(t, testStruct, actualStruct)
+ assert.Equal(t, "1985-06-02T00:00:00Z", testStruct.Day.String())
}
// Test pointers of DayValue.
@@ -68,35 +68,6 @@ func TestDayValueYAML(t *testing.T) {
err = yaml.Unmarshal(expected, &actualStruct)
require.NoError(t, err)
assert.Equal(t, testStruct, actualStruct)
- }
- // Test UTC-stable string and YAML serialization in western timezones.
- {
- loc, err := time.LoadLocation("America/Los_Angeles")
- if err != nil {
- loc = time.FixedZone("UTC-8", -8*60*60)
- }
-
- originalLocal := time.Local
- time.Local = loc
- defer func() {
- time.Local = originalLocal
- }()
- type TestStruct struct {
- Day *DayValue `yaml:"day"`
- }
- var testStruct TestStruct
- testStruct.Day = &DayValue{}
- require.NoError(t, testStruct.Day.Set("1985-06-02"))
- expected := []byte(`day: "1985-06-02"
-`)
-
- actual, err := yaml.Marshal(testStruct)
- require.NoError(t, err)
- assert.Equal(t, expected, actual)
-
- var actualStruct TestStruct
- err = yaml.Unmarshal(expected, &actualStruct)
- require.NoError(t, err)
- assert.Equal(t, testStruct, actualStruct)
+ assert.Equal(t, "1985-06-02T00:00:00Z", testStruct.Day.String())
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
