You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@seata.apache.org by te...@apache.org on 2026/10/24 08:38:14 UTC
(incubator-seata-go) branch master updated: fix(rm): send branch failure response when branch operation fails (#1164)
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 23b9edd9 fix(rm): send branch failure response when branch operation fails (#1164)
23b9edd9 is described below
commit 23b9edd9e43c5f13f64206acc20bbffb2e6f0122
Author: XiaoFei <20...@qq.com>
AuthorDate: Sat Oct 24 16:38:10 2026 +0800
fix(rm): send branch failure response when branch operation fails (#1164)
* fix: send branch failure response when RM operation fails
* fix: prevent branch rollback response message length overflow
* fix(rm): address branch failure response review feedback
* fix(rm): add missing branch end processor helpers
* fix(rm): complete branch failure response review fixes
---------
Co-authored-by: ThunGuo <te...@apache.org>
---
changes/dev.md | 4 +-
pkg/protocol/codec/branch_commit_response_codec.go | 8 +-
.../codec/branch_commit_response_codec_test.go | 62 ++++
.../codec/branch_rollback_response_codec.go | 4 +-
.../codec/branch_rollback_response_codec_test.go | 62 ++++
.../processor/client/rm_branch_commit_processor.go | 83 ++---
.../client/rm_branch_commit_processor_test.go | 349 ++++++++++++++-------
.../processor/client/rm_branch_end_processor.go | 80 +++++
.../client/rm_branch_rollback_processor.go | 87 +++--
.../client/rm_branch_rollback_processor_test.go | 258 ++++++++-------
10 files changed, 665 insertions(+), 332 deletions(-)
diff --git a/changes/dev.md b/changes/dev.md
index 8f97e0bd..197efade 100755
--- a/changes/dev.md
+++ b/changes/dev.md
@@ -32,11 +32,9 @@
### bugfix：
+ - [[#1156](https://github.com/apache/incubator-seata-go/issues/1156)] send RM branch failure responses to TC and align their message encoding with the Java wire format
- [[#1190](https://github.com/apache/incubator-seata-go/issues/1190)] fix UPDATE JOIN after-image queries to use primary keys from before images and correctly handle unmatched outer-join targets
- [[#1207](https://github.com/apache/incubator-seata-go/pull/1207)] fix metadata cache overwrites across multiple data sources of the same database type and retain schema-qualified PostgreSQL table identities.
-
- **action required:** PostgreSQL AT global lock keys now include the schema, for example `ACCOUNT:1` becomes `PUBLIC.ACCOUNT:1`. The old and new formats are incompatible. Before upgrading, pause incoming requests, message consumers, and business scheduled jobs. Keep the current RMs, TC, and recovery components online until all affected global transactions and branches have committed or rolled back, including phase-two retries and undo-log cleanup, and all related global locks have bee [...]
-
- [[#1201](https://github.com/apache/incubator-seata-go/issues/1201)] report a retryable TCC branch failure when a phase-two action returns false
- [[#1169](https://github.com/apache/incubator-seata-go/issues/1169)] align gRPC protobuf message types with the Seata protocol
- [[#1165](https://github.com/apache/incubator-seata-go/issues/1165)] fix incorrect gRPC global commit request message type
diff --git a/pkg/protocol/codec/branch_commit_response_codec.go b/pkg/protocol/codec/branch_commit_response_codec.go
index 047382a0..600fda0e 100644
--- a/pkg/protocol/codec/branch_commit_response_codec.go
+++ b/pkg/protocol/codec/branch_commit_response_codec.go
@@ -35,7 +35,7 @@ func (g *BranchCommitResponseCodec) Decode(in []byte) interface{} {
data.ResultCode = message.ResultCode(bytes.ReadByte(buf))
if data.ResultCode == message.ResultCodeFailed {
- data.Msg = bytes.ReadString8Length(buf)
+ data.Msg = bytes.ReadString16Length(buf)
}
data.TransactionErrorCode = serror.TransactionErrorCode(bytes.ReadByte(buf))
data.Xid = bytes.ReadString16Length(buf)
@@ -52,10 +52,10 @@ func (g *BranchCommitResponseCodec) Encode(in interface{}) []byte {
buf.WriteByte(byte(data.ResultCode))
if data.ResultCode == message.ResultCodeFailed {
msg := data.Msg
- if len(data.Msg) > math.MaxInt8 {
- msg = data.Msg[:math.MaxInt8]
+ if len(data.Msg) > math.MaxInt16 {
+ msg = data.Msg[:math.MaxInt16]
}
- bytes.WriteString8Length(msg, buf)
+ bytes.WriteString16Length(msg, buf)
}
buf.WriteByte(byte(data.TransactionErrorCode))
bytes.WriteString16Length(data.Xid, buf)
diff --git a/pkg/protocol/codec/branch_commit_response_codec_test.go b/pkg/protocol/codec/branch_commit_response_codec_test.go
index 7656b3c9..37308e5f 100644
--- a/pkg/protocol/codec/branch_commit_response_codec_test.go
+++ b/pkg/protocol/codec/branch_commit_response_codec_test.go
@@ -18,6 +18,8 @@
package codec
import (
+ "math"
+ "strings"
"testing"
"github.com/stretchr/testify/assert"
@@ -49,3 +51,63 @@ func TestBranchCommitResponseCodec(t *testing.T) {
assert.Equal(t, msg, msg2)
}
+
+func TestBranchCommitResponseCodec_TruncatesLongMessageWithoutShiftingFields(t *testing.T) {
+ msg := message.BranchCommitResponse{
+ AbstractBranchEndResponse: message.AbstractBranchEndResponse{
+ Xid: "xid",
+ BranchId: 123,
+ BranchStatus: model2.BranchStatusPhasetwoCommitFailedRetryable,
+ AbstractTransactionResponse: message.AbstractTransactionResponse{
+ TransactionErrorCode: serror.TransactionErrorCodeFailedToSendBranchCommitRequest,
+ AbstractResultMessage: message.AbstractResultMessage{
+ ResultCode: message.ResultCodeFailed,
+ Msg: strings.Repeat("x", math.MaxInt16+100),
+ },
+ },
+ },
+ }
+
+ codec := BranchCommitResponseCodec{}
+ decoded := codec.Decode(codec.Encode(msg)).(message.BranchCommitResponse)
+
+ assert.Equal(t, message.ResultCodeFailed, decoded.ResultCode)
+ assert.Equal(t, strings.Repeat("x", math.MaxInt16), decoded.Msg)
+ assert.Equal(t, msg.TransactionErrorCode, decoded.TransactionErrorCode)
+ assert.Equal(t, msg.Xid, decoded.Xid)
+ assert.Equal(t, msg.BranchId, decoded.BranchId)
+ assert.Equal(t, msg.BranchStatus, decoded.BranchStatus)
+}
+
+// Byte vector derived from Java AbstractResultMessageCodec,
+// AbstractTransactionResponseCodec, and AbstractBranchEndResponseCodec.
+func TestBranchCommitResponseCodec_JavaWireFormat(t *testing.T) {
+ msg := message.BranchCommitResponse{
+ AbstractBranchEndResponse: message.AbstractBranchEndResponse{
+ Xid: "192.168.0.1:8091:1234",
+ BranchId: 5678,
+ BranchStatus: model2.BranchStatusPhasetwoCommitFailedRetryable,
+ AbstractTransactionResponse: message.AbstractTransactionResponse{
+ TransactionErrorCode: serror.TransactionErrorCodeUnknown,
+ AbstractResultMessage: message.AbstractResultMessage{
+ ResultCode: message.ResultCodeFailed,
+ Msg: "storage failed",
+ },
+ },
+ },
+ }
+ want := []byte{
+ 0x00,
+ 0x00, 0x0e,
+ 's', 't', 'o', 'r', 'a', 'g', 'e', ' ', 'f', 'a', 'i', 'l', 'e', 'd',
+ 0x00,
+ 0x00, 0x15,
+ '1', '9', '2', '.', '1', '6', '8', '.', '0', '.', '1', ':', '8', '0', '9', '1', ':', '1', '2', '3', '4',
+ 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x16, 0x2e,
+ 0x06,
+ }
+
+ codec := BranchCommitResponseCodec{}
+ assert.Equal(t, want, codec.Encode(msg), "encode must reproduce the Java bytes exactly")
+ assert.Equal(t, msg, codec.Decode(want))
+}
diff --git a/pkg/protocol/codec/branch_rollback_response_codec.go b/pkg/protocol/codec/branch_rollback_response_codec.go
index 1c627f93..f157fa3e 100644
--- a/pkg/protocol/codec/branch_rollback_response_codec.go
+++ b/pkg/protocol/codec/branch_rollback_response_codec.go
@@ -34,7 +34,7 @@ func (g *BranchRollbackResponseCodec) Decode(in []byte) interface{} {
data.ResultCode = message.ResultCode(bytes.ReadByte(buf))
if data.ResultCode == message.ResultCodeFailed {
- data.Msg = bytes.ReadString8Length(buf)
+ data.Msg = bytes.ReadString16Length(buf)
}
data.TransactionErrorCode = serror.TransactionErrorCode(bytes.ReadByte(buf))
data.Xid = bytes.ReadString16Length(buf)
@@ -54,7 +54,7 @@ func (g *BranchRollbackResponseCodec) Encode(in interface{}) []byte {
if len(data.Msg) > math.MaxInt16 {
msg = data.Msg[:math.MaxInt16]
}
- bytes.WriteString8Length(msg, buf)
+ bytes.WriteString16Length(msg, buf)
}
buf.WriteByte(byte(data.TransactionErrorCode))
bytes.WriteString16Length(data.Xid, buf)
diff --git a/pkg/protocol/codec/branch_rollback_response_codec_test.go b/pkg/protocol/codec/branch_rollback_response_codec_test.go
index 0c99a829..8e17e312 100644
--- a/pkg/protocol/codec/branch_rollback_response_codec_test.go
+++ b/pkg/protocol/codec/branch_rollback_response_codec_test.go
@@ -18,6 +18,8 @@
package codec
import (
+ "math"
+ "strings"
"testing"
serror "seata.apache.org/seata-go/v2/pkg/util/errors"
@@ -51,3 +53,63 @@ func TestBranchRollbackResponseCodec(t *testing.T) {
assert.Equal(t, msg, msg2)
}
+
+func TestBranchRollbackResponseCodec_TruncatesLongMessageWithoutShiftingFields(t *testing.T) {
+ msg := message.BranchRollbackResponse{
+ AbstractBranchEndResponse: message.AbstractBranchEndResponse{
+ Xid: "xid",
+ BranchId: 123,
+ BranchStatus: model2.BranchStatusPhasetwoRollbackFailedRetryable,
+ AbstractTransactionResponse: message.AbstractTransactionResponse{
+ TransactionErrorCode: serror.TransactionErrorCodeBranchRollbackFailedRetriable,
+ AbstractResultMessage: message.AbstractResultMessage{
+ ResultCode: message.ResultCodeFailed,
+ Msg: strings.Repeat("x", math.MaxInt16+100),
+ },
+ },
+ },
+ }
+
+ codec := BranchRollbackResponseCodec{}
+ decoded := codec.Decode(codec.Encode(msg)).(message.BranchRollbackResponse)
+
+ assert.Equal(t, message.ResultCodeFailed, decoded.ResultCode)
+ assert.Equal(t, strings.Repeat("x", math.MaxInt16), decoded.Msg)
+ assert.Equal(t, msg.TransactionErrorCode, decoded.TransactionErrorCode)
+ assert.Equal(t, msg.Xid, decoded.Xid)
+ assert.Equal(t, msg.BranchId, decoded.BranchId)
+ assert.Equal(t, msg.BranchStatus, decoded.BranchStatus)
+}
+
+// Byte vector derived from Java AbstractResultMessageCodec,
+// AbstractTransactionResponseCodec, and AbstractBranchEndResponseCodec.
+func TestBranchRollbackResponseCodec_JavaWireFormat(t *testing.T) {
+ msg := message.BranchRollbackResponse{
+ AbstractBranchEndResponse: message.AbstractBranchEndResponse{
+ Xid: "192.168.0.1:8091:1234",
+ BranchId: 5678,
+ BranchStatus: model2.BranchStatusPhasetwoRollbackFailedRetryable,
+ AbstractTransactionResponse: message.AbstractTransactionResponse{
+ TransactionErrorCode: serror.TransactionErrorCodeBranchRollbackFailedRetriable,
+ AbstractResultMessage: message.AbstractResultMessage{
+ ResultCode: message.ResultCodeFailed,
+ Msg: "storage failed",
+ },
+ },
+ },
+ }
+ want := []byte{
+ 0x00,
+ 0x00, 0x0e,
+ 's', 't', 'o', 'r', 'a', 'g', 'e', ' ', 'f', 'a', 'i', 'l', 'e', 'd',
+ 0x04,
+ 0x00, 0x15,
+ '1', '9', '2', '.', '1', '6', '8', '.', '0', '.', '1', ':', '8', '0', '9', '1', ':', '1', '2', '3', '4',
+ 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x16, 0x2e,
+ 0x09,
+ }
+
+ codec := BranchRollbackResponseCodec{}
+ assert.Equal(t, want, codec.Encode(msg), "encode must reproduce the Java bytes exactly")
+ assert.Equal(t, msg, codec.Decode(want))
+}
diff --git a/pkg/remoting/processor/client/rm_branch_commit_processor.go b/pkg/remoting/processor/client/rm_branch_commit_processor.go
index fde54ab4..6e95ef5a 100644
--- a/pkg/remoting/processor/client/rm_branch_commit_processor.go
+++ b/pkg/remoting/processor/client/rm_branch_commit_processor.go
@@ -45,7 +45,10 @@ func initBranchCommit() {
}
}
-type rmBranchCommitProcessor struct{}
+type rmBranchCommitProcessor struct {
+ getResourceManager func(branch.BranchType) rm.ResourceManager
+ sendResponse func(int32, interface{}) error
+}
func (f *rmBranchCommitProcessor) Process(ctx context.Context, rpcMessage message.RpcMessage) error {
log.Infof("the rm client received rmBranchCommit rpcMessage %#v from tc server.", rpcMessage)
@@ -71,23 +74,13 @@ func (f *rmBranchCommitProcessor) handleGrpcBranchCommit(ctx context.Context, rp
Xid: xid,
}
- status, err := rm.GetRmCacheInstance().GetResourceManager(branch.BranchType(request.AbstractBranchEndRequest.BranchType)).BranchCommit(ctx, branchResource)
- if err != nil {
- log.Errorf("branch commit error: %s", err.Error())
- return err
- }
- log.Infof("branch commit success: xid %s, branchID %d, resourceID %s, applicationData %s", xid, branchID, resourceID, applicationData)
-
- var (
- resultCode pb.ResultCodeProto
- errMsg string
- )
- if err != nil {
- resultCode = pb.ResultCodeProto_Failed
- errMsg = err.Error()
+ status, bizErr := branchEndResourceManager(f.getResourceManager, branch.BranchType(request.AbstractBranchEndRequest.BranchType)).BranchCommit(ctx, branchResource)
+ if bizErr != nil {
+ log.Errorf("branch commit error: xid %s, branchID %d: %s", xid, branchID, bizErr.Error())
} else {
- resultCode = pb.ResultCodeProto_Success
+ log.Infof("branch commit success: xid %s, branchID %d, resourceID %s, applicationData %s", xid, branchID, resourceID, applicationData)
}
+ result := newBranchEndResult(status, bizErr)
// reply commit response to tc server
// todo add TransactionErrorCode
@@ -96,23 +89,24 @@ func (f *rmBranchCommitProcessor) handleGrpcBranchCommit(ctx context.Context, rp
AbstractTransactionResponse: &pb.AbstractTransactionResponseProto{
AbstractResultMessage: &pb.AbstractResultMessageProto{
AbstractMessage: &pb.AbstractMessageProto{MessageType: pb.MessageTypeProto_TYPE_BRANCH_COMMIT_RESULT},
- ResultCode: resultCode,
- Msg: errMsg,
+ ResultCode: branchEndResultCodeProto(result.resultCode),
+ Msg: result.errMsg,
},
},
Xid: xid,
BranchId: branchID,
- BranchStatus: pb.BranchStatusProto(status),
+ BranchStatus: pb.BranchStatusProto(result.status),
},
}
- err = grpc.GetGrpcRemotingClient().SendAsyncResponse(rpcMessage.ID, response)
- if err != nil {
- log.Errorf("send branch commit response error: {%#v}", err.Error())
- return err
+ sendResponse := branchEndSendResponse(f.sendResponse, grpc.GetGrpcRemotingClient().SendAsyncResponse)
+ sendErr := sendResponse(rpcMessage.ID, response)
+ if sendErr != nil {
+ log.Errorf("send branch commit response error: xid %s, branchID %d: %v", xid, branchID, sendErr)
+ } else {
+ log.Infof("send branch commit response success: xid %s, branchID %v, resourceID %v, applicationData %v", xid, branchID, resourceID, applicationData)
}
- log.Infof("send branch commit success: xid %v, branchID %v, resourceID %v, applicationData %v", xid, branchID, resourceID, applicationData)
- return nil
+ return branchEndProcessError(bizErr, sendErr)
}
func (f *rmBranchCommitProcessor) handleGettyBranchCommit(ctx context.Context, rpcMessage message.RpcMessage) error {
@@ -129,23 +123,13 @@ func (f *rmBranchCommitProcessor) handleGettyBranchCommit(ctx context.Context, r
Xid: xid,
}
- status, err := rm.GetRmCacheInstance().GetResourceManager(request.BranchType).BranchCommit(ctx, branchResource)
- if err != nil {
- log.Errorf("branch commit error: %s", err.Error())
- return err
- }
- log.Infof("branch commit success: xid %s, branchID %d, resourceID %s, applicationData %s", xid, branchID, resourceID, applicationData)
-
- var (
- resultCode message.ResultCode
- errMsg string
- )
- if err != nil {
- resultCode = message.ResultCodeFailed
- errMsg = err.Error()
+ status, bizErr := branchEndResourceManager(f.getResourceManager, request.BranchType).BranchCommit(ctx, branchResource)
+ if bizErr != nil {
+ log.Errorf("branch commit error: xid %s, branchID %d: %s", xid, branchID, bizErr.Error())
} else {
- resultCode = message.ResultCodeSuccess
+ log.Infof("branch commit success: xid %s, branchID %d, resourceID %s, applicationData %s", xid, branchID, resourceID, applicationData)
}
+ result := newBranchEndResult(status, bizErr)
// reply commit response to tc server
// todo add TransactionErrorCode
@@ -153,20 +137,21 @@ func (f *rmBranchCommitProcessor) handleGettyBranchCommit(ctx context.Context, r
AbstractBranchEndResponse: message.AbstractBranchEndResponse{
AbstractTransactionResponse: message.AbstractTransactionResponse{
AbstractResultMessage: message.AbstractResultMessage{
- ResultCode: resultCode,
- Msg: errMsg,
+ ResultCode: result.resultCode,
+ Msg: result.errMsg,
},
},
Xid: xid,
BranchId: branchID,
- BranchStatus: status,
+ BranchStatus: result.status,
},
}
- err = getty.GetGettyRemotingClient().SendAsyncResponse(rpcMessage.ID, response)
- if err != nil {
- log.Errorf("send branch commit response error: {%#v}", err.Error())
- return err
+ sendResponse := branchEndSendResponse(f.sendResponse, getty.GetGettyRemotingClient().SendAsyncResponse)
+ sendErr := sendResponse(rpcMessage.ID, response)
+ if sendErr != nil {
+ log.Errorf("send branch commit response error: xid %s, branchID %d: %v", xid, branchID, sendErr)
+ } else {
+ log.Infof("send branch commit response success: xid %s, branchID %v, resourceID %v, applicationData %v", xid, branchID, resourceID, applicationData)
}
- log.Infof("send branch commit success: xid %v, branchID %v, resourceID %v, applicationData %v", xid, branchID, resourceID, applicationData)
- return nil
+ return branchEndProcessError(bizErr, sendErr)
}
diff --git a/pkg/remoting/processor/client/rm_branch_commit_processor_test.go b/pkg/remoting/processor/client/rm_branch_commit_processor_test.go
index 5c41c27e..ae5d7eaf 100644
--- a/pkg/remoting/processor/client/rm_branch_commit_processor_test.go
+++ b/pkg/remoting/processor/client/rm_branch_commit_processor_test.go
@@ -19,146 +19,259 @@ package client
import (
"context"
- "reflect"
+ "errors"
+ "sync"
"testing"
- "github.com/agiledragon/gomonkey/v2"
- "github.com/stretchr/testify/assert"
-
- "seata.apache.org/seata-go/v2/pkg/rm/tcc"
-
- model2 "seata.apache.org/seata-go/v2/pkg/protocol/branch"
- "seata.apache.org/seata-go/v2/pkg/protocol/codec"
+ "github.com/stretchr/testify/require"
+ "seata.apache.org/seata-go/v2/pkg/protocol"
+ "seata.apache.org/seata-go/v2/pkg/protocol/branch"
"seata.apache.org/seata-go/v2/pkg/protocol/message"
"seata.apache.org/seata-go/v2/pkg/remoting/config"
- remotinggrpc "seata.apache.org/seata-go/v2/pkg/remoting/grpc"
"seata.apache.org/seata-go/v2/pkg/remoting/grpc/pb"
"seata.apache.org/seata-go/v2/pkg/rm"
)
-func TestRmBranchCommitProcessor(t *testing.T) {
- // testcases
- tests := []struct {
- name string // testcase name
- protocol string // protocol:seata/grpc
- rpcMsg message.RpcMessage // rpcMessage case
- wantErr bool // want testcase err or not
- }{
- {
- name: "rbc-testcase1-failure",
- protocol: "seata",
- rpcMsg: message.RpcMessage{
- ID: 123,
- Type: message.RequestType(message.MessageTypeBranchCommit),
- Codec: byte(codec.CodecTypeSeata),
- Compressor: byte(1),
- HeadMap: map[string]string{
- "name": " Jack",
- "age": "12",
- "address": "Beijing",
- },
- Body: message.BranchCommitRequest{
- AbstractBranchEndRequest: message.AbstractBranchEndRequest{
- Xid: "123344",
- BranchId: 56678,
- BranchType: model2.BranchTypeTCC,
- ResourceId: "1232323",
- ApplicationData: []byte("TestExtraData"),
- },
- },
- },
-
- wantErr: true, // need dail to server, so err accured
- },
- {
- name: "rbc-testcase2-failure",
- protocol: "grpc",
- rpcMsg: message.RpcMessage{
- ID: 123,
- Type: message.RequestType(message.MessageTypeBranchCommit),
- HeadMap: map[string]string{
- "name": " Jack",
- "age": "12",
- "address": "Beijing",
- },
- Body: &pb.BranchCommitRequestProto{
- AbstractBranchEndRequest: &pb.AbstractBranchEndRequestProto{
- Xid: "123345",
- BranchId: 56679,
- BranchType: pb.BranchTypeProto_TCC,
- ResourceId: "1232324",
- ApplicationData: "TestExtraData",
- },
- },
- },
-
- wantErr: true, // need dail to server, so err accured
- },
+type testResourceManager struct {
+ branchType branch.BranchType
+ commitStatus branch.BranchStatus
+ commitErr error
+ rollbackStatus branch.BranchStatus
+ rollbackErr error
+}
+
+func (m *testResourceManager) BranchCommit(context.Context, rm.BranchResource) (branch.BranchStatus, error) {
+ return m.commitStatus, m.commitErr
+}
+func (m *testResourceManager) BranchRollback(context.Context, rm.BranchResource) (branch.BranchStatus, error) {
+ return m.rollbackStatus, m.rollbackErr
+}
+func (*testResourceManager) BranchRegister(context.Context, rm.BranchRegisterParam) (int64, error) {
+ return 0, nil
+}
+func (*testResourceManager) BranchReport(context.Context, rm.BranchReportParam) error { return nil }
+func (*testResourceManager) LockQuery(context.Context, rm.LockQueryParam) (bool, error) {
+ return false, nil
+}
+func (*testResourceManager) RegisterResource(rm.Resource) error { return nil }
+func (*testResourceManager) UnregisterResource(rm.Resource) error { return nil }
+func (*testResourceManager) GetCachedResources() *sync.Map { return &sync.Map{} }
+func (m *testResourceManager) GetBranchType() branch.BranchType { return m.branchType }
+
+func TestBranchEndSendResponse(t *testing.T) {
+ injectedCalled := false
+ fallbackCalled := false
+ injected := func(int32, interface{}) error {
+ injectedCalled = true
+ return nil
+ }
+ fallback := func(int32, interface{}) error {
+ fallbackCalled = true
+ return nil
}
- var ctx context.Context
- var rbcProcessor rmBranchCommitProcessor
+ require.NoError(t, branchEndSendResponse(injected, fallback)(1, "response"))
+ require.True(t, injectedCalled)
+ require.False(t, fallbackCalled)
- // run tests
- for _, tc := range tests {
- t.Run(tc.name, func(t *testing.T) {
- config.InitTransportConfig(&config.TransportConfig{Protocol: tc.protocol})
+ injectedCalled = false
+ require.NoError(t, branchEndSendResponse(nil, fallback)(2, "response"))
+ require.False(t, injectedCalled)
+ require.True(t, fallbackCalled)
+}
- rm.GetRmCacheInstance().RegisterResourceManager(tcc.GetTCCResourceManagerInstance())
+func TestBranchEndResult(t *testing.T) {
+ bizErr := errors.New("operation failed")
+ failed := newBranchEndResult(branch.BranchStatusPhasetwoCommitFailedRetryable, bizErr)
+ require.Equal(t, message.ResultCodeFailed, failed.resultCode)
+ require.Equal(t, pb.ResultCodeProto_Failed, branchEndResultCodeProto(failed.resultCode))
+ require.Equal(t, bizErr.Error(), failed.errMsg)
- err := rbcProcessor.Process(ctx, tc.rpcMsg)
- if (err != nil) != tc.wantErr {
- t.Errorf("rmBranchCommitProcessor wantErr: %v, got: %v", tc.wantErr, err)
- return
- }
- })
- }
+ success := newBranchEndResult(branch.BranchStatusPhasetwoCommitted, nil)
+ require.Equal(t, message.ResultCodeSuccess, success.resultCode)
+ require.Equal(t, pb.ResultCodeProto_Success, branchEndResultCodeProto(success.resultCode))
+ require.Empty(t, success.errMsg)
+
+ sendErr := errors.New("send failed")
+ require.NoError(t, branchEndProcessError(bizErr, nil))
+ require.ErrorIs(t, branchEndProcessError(nil, sendErr), sendErr)
+ combined := branchEndProcessError(bizErr, sendErr)
+ require.ErrorIs(t, combined, bizErr)
+ require.ErrorIs(t, combined, sendErr)
+
+ injectedManager := &testResourceManager{}
+ require.Same(t, injectedManager, branchEndResourceManager(
+ func(branch.BranchType) rm.ResourceManager { return injectedManager },
+ branch.BranchTypeTCC,
+ ))
+
+ fallbackType := branch.BranchType(99)
+ fallbackManager := &testResourceManager{branchType: fallbackType}
+ rmCache := rm.GetRmCacheInstance()
+ rmCache.RegisterResourceManager(fallbackManager)
+ t.Cleanup(func() { rmCache.UnregisterResourceManager(fallbackType) })
+ require.Same(t, fallbackManager, branchEndResourceManager(nil, fallbackType))
}
-func TestRmBranchCommitProcessorResponseUsesCommitResultMessageType(t *testing.T) {
- resourceManager := &testGrpcResourceManager{branchType: model2.BranchTypeTCC}
- rm.GetRmCacheInstance().RegisterResourceManager(resourceManager)
- defer rm.GetRmCacheInstance().UnregisterResourceManager(model2.BranchTypeTCC)
- config.InitTransportConfig(&config.TransportConfig{Protocol: "grpc"})
-
- var response *pb.BranchCommitResponseProto
- patches := gomonkey.ApplyMethod(reflect.TypeOf(remotinggrpc.GetGrpcRemotingClient()), "SendAsyncResponse",
- func(_ *remotinggrpc.GrpcRemotingClient, _ int32, msg interface{}) error {
- response = msg.(*pb.BranchCommitResponseProto)
- return nil
- })
- defer patches.Reset()
-
- err := (&rmBranchCommitProcessor{}).Process(context.Background(), message.RpcMessage{
- ID: 1,
- Type: message.RequestType(message.MessageTypeBranchCommit),
- Body: &pb.BranchCommitRequestProto{AbstractBranchEndRequest: &pb.AbstractBranchEndRequestProto{
- Xid: "test-xid",
- BranchId: 1,
- BranchType: pb.BranchTypeProto_TCC,
- }},
+func TestRmBranchCommitProcessor_SendsFailureResponse(t *testing.T) {
+ bizErr := errors.New("commit failed")
+ manager := &testResourceManager{commitStatus: branch.BranchStatusPhasetwoCommitFailedRetryable, commitErr: bizErr}
+
+ t.Run("getty", func(t *testing.T) {
+ var sent interface{}
+ processor := rmBranchCommitProcessor{
+ getResourceManager: func(branch.BranchType) rm.ResourceManager { return manager },
+ sendResponse: func(_ int32, response interface{}) error { sent = response; return nil },
+ }
+ err := processor.handleGettyBranchCommit(context.Background(), message.RpcMessage{ID: 1, Body: message.BranchCommitRequest{
+ AbstractBranchEndRequest: message.AbstractBranchEndRequest{Xid: "xid", BranchId: 7, BranchType: branch.BranchTypeTCC, ResourceId: "resource"},
+ }})
+ require.NoError(t, err)
+ got := sent.(message.BranchCommitResponse)
+ require.Equal(t, message.ResultCodeFailed, got.ResultCode)
+ require.Equal(t, bizErr.Error(), got.Msg)
+ require.Equal(t, manager.commitStatus, got.BranchStatus)
+ require.Equal(t, "xid", got.Xid)
+ require.Equal(t, int64(7), got.BranchId)
})
- assert.NoError(t, err)
- if assert.NotNil(t, response) {
- assert.Equal(t, pb.MessageTypeProto_TYPE_BRANCH_COMMIT_RESULT,
- response.GetAbstractBranchEndResponse().GetAbstractTransactionResponse().GetAbstractResultMessage().GetAbstractMessage().GetMessageType())
- }
+ t.Run("grpc", func(t *testing.T) {
+ var sent interface{}
+ processor := rmBranchCommitProcessor{
+ getResourceManager: func(branch.BranchType) rm.ResourceManager { return manager },
+ sendResponse: func(_ int32, response interface{}) error { sent = response; return nil },
+ }
+ err := processor.handleGrpcBranchCommit(context.Background(), message.RpcMessage{ID: 1, Body: &pb.BranchCommitRequestProto{
+ AbstractBranchEndRequest: &pb.AbstractBranchEndRequestProto{Xid: "xid", BranchId: 7, BranchType: pb.BranchTypeProto_TCC, ResourceId: "resource"},
+ }})
+ require.NoError(t, err)
+ got := sent.(*pb.BranchCommitResponseProto)
+ result := got.AbstractBranchEndResponse.AbstractTransactionResponse.AbstractResultMessage
+ require.Equal(t, pb.MessageTypeProto_TYPE_BRANCH_COMMIT_RESULT, result.GetAbstractMessage().GetMessageType())
+ require.Equal(t, pb.ResultCodeProto_Failed, result.ResultCode)
+ require.Equal(t, bizErr.Error(), result.Msg)
+ require.Equal(t, pb.BranchStatusProto(manager.commitStatus), got.AbstractBranchEndResponse.BranchStatus)
+ require.Equal(t, "xid", got.AbstractBranchEndResponse.Xid)
+ require.Equal(t, int64(7), got.AbstractBranchEndResponse.BranchId)
+ })
}
-type testGrpcResourceManager struct {
- rm.ResourceManager
- branchType model2.BranchType
-}
+func TestRmBranchCommitProcessor_ObservesBusinessAndSendErrors(t *testing.T) {
+ bizErr := errors.New("commit failed")
+ sendErr := errors.New("send failed")
+ manager := &testResourceManager{commitStatus: branch.BranchStatusPhasetwoCommitFailedRetryable, commitErr: bizErr}
-func (m *testGrpcResourceManager) GetBranchType() model2.BranchType {
- return m.branchType
+ t.Run("getty", func(t *testing.T) {
+ processor := rmBranchCommitProcessor{
+ getResourceManager: func(branch.BranchType) rm.ResourceManager { return manager },
+ sendResponse: func(int32, interface{}) error { return sendErr },
+ }
+ err := processor.handleGettyBranchCommit(context.Background(), message.RpcMessage{ID: 1, Body: message.BranchCommitRequest{
+ AbstractBranchEndRequest: message.AbstractBranchEndRequest{Xid: "xid", BranchId: 7, BranchType: branch.BranchTypeTCC, ResourceId: "resource"},
+ }})
+ require.ErrorIs(t, err, bizErr)
+ require.ErrorIs(t, err, sendErr)
+ })
+
+ t.Run("grpc", func(t *testing.T) {
+ processor := rmBranchCommitProcessor{
+ getResourceManager: func(branch.BranchType) rm.ResourceManager { return manager },
+ sendResponse: func(int32, interface{}) error { return sendErr },
+ }
+ err := processor.handleGrpcBranchCommit(context.Background(), message.RpcMessage{ID: 1, Body: &pb.BranchCommitRequestProto{
+ AbstractBranchEndRequest: &pb.AbstractBranchEndRequestProto{Xid: "xid", BranchId: 7, BranchType: pb.BranchTypeProto_TCC, ResourceId: "resource"},
+ }})
+ require.ErrorIs(t, err, bizErr)
+ require.ErrorIs(t, err, sendErr)
+ })
}
-func (m *testGrpcResourceManager) BranchCommit(context.Context, rm.BranchResource) (model2.BranchStatus, error) {
- return model2.BranchStatusPhaseoneDone, nil
+func TestRmBranchCommitProcessor_SendsSuccessResponse(t *testing.T) {
+ manager := &testResourceManager{commitStatus: branch.BranchStatusPhasetwoCommitted}
+
+ t.Run("getty", func(t *testing.T) {
+ var sent interface{}
+ processor := rmBranchCommitProcessor{
+ getResourceManager: func(branch.BranchType) rm.ResourceManager { return manager },
+ sendResponse: func(_ int32, response interface{}) error { sent = response; return nil },
+ }
+ err := processor.handleGettyBranchCommit(context.Background(), message.RpcMessage{ID: 1, Body: message.BranchCommitRequest{
+ AbstractBranchEndRequest: message.AbstractBranchEndRequest{Xid: "xid", BranchId: 7, BranchType: branch.BranchTypeTCC, ResourceId: "resource"},
+ }})
+ require.NoError(t, err)
+ got := sent.(message.BranchCommitResponse)
+ require.Equal(t, message.ResultCodeSuccess, got.ResultCode)
+ require.Empty(t, got.Msg)
+ require.Equal(t, manager.commitStatus, got.BranchStatus)
+ require.Equal(t, "xid", got.Xid)
+ require.Equal(t, int64(7), got.BranchId)
+ })
+
+ t.Run("grpc", func(t *testing.T) {
+ var sent interface{}
+ processor := rmBranchCommitProcessor{
+ getResourceManager: func(branch.BranchType) rm.ResourceManager { return manager },
+ sendResponse: func(_ int32, response interface{}) error { sent = response; return nil },
+ }
+ err := processor.handleGrpcBranchCommit(context.Background(), message.RpcMessage{ID: 1, Body: &pb.BranchCommitRequestProto{
+ AbstractBranchEndRequest: &pb.AbstractBranchEndRequestProto{Xid: "xid", BranchId: 7, BranchType: pb.BranchTypeProto_TCC, ResourceId: "resource"},
+ }})
+ require.NoError(t, err)
+ got := sent.(*pb.BranchCommitResponseProto)
+ result := got.AbstractBranchEndResponse.AbstractTransactionResponse.AbstractResultMessage
+ require.Equal(t, pb.MessageTypeProto_TYPE_BRANCH_COMMIT_RESULT, result.GetAbstractMessage().GetMessageType())
+ require.Equal(t, pb.ResultCodeProto_Success, result.ResultCode)
+ require.Empty(t, result.Msg)
+ require.Equal(t, pb.BranchStatusProto(manager.commitStatus), got.AbstractBranchEndResponse.BranchStatus)
+ require.Equal(t, "xid", got.AbstractBranchEndResponse.Xid)
+ require.Equal(t, int64(7), got.AbstractBranchEndResponse.BranchId)
+ })
}
-func (m *testGrpcResourceManager) BranchRollback(context.Context, rm.BranchResource) (model2.BranchStatus, error) {
- return model2.BranchStatusPhaseoneDone, nil
+func TestRmBranchCommitProcessor_ProcessRoutesByProtocol(t *testing.T) {
+ previous := config.GetTransportConfig()
+ t.Cleanup(func() { config.InitTransportConfig(previous) })
+
+ manager := &testResourceManager{commitStatus: branch.BranchStatusPhasetwoCommitted}
+ tests := []struct {
+ name string
+ protocol protocol.Protocol
+ body interface{}
+ wantType interface{}
+ }{
+ {
+ name: "seata",
+ protocol: protocol.ProtocolSEATA,
+ body: message.BranchCommitRequest{AbstractBranchEndRequest: message.AbstractBranchEndRequest{
+ Xid: "xid", BranchId: 7, BranchType: branch.BranchTypeTCC, ResourceId: "resource",
+ }},
+ wantType: message.BranchCommitResponse{},
+ },
+ {
+ name: "grpc",
+ protocol: protocol.ProtocolGRPC,
+ body: &pb.BranchCommitRequestProto{AbstractBranchEndRequest: &pb.AbstractBranchEndRequestProto{
+ Xid: "xid", BranchId: 7, BranchType: pb.BranchTypeProto_TCC, ResourceId: "resource",
+ }},
+ wantType: (*pb.BranchCommitResponseProto)(nil),
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ config.InitTransportConfig(&config.TransportConfig{Protocol: tt.protocol.String()})
+ var sent interface{}
+ processor := rmBranchCommitProcessor{
+ getResourceManager: func(branch.BranchType) rm.ResourceManager { return manager },
+ sendResponse: func(_ int32, response interface{}) error { sent = response; return nil },
+ }
+
+ require.NoError(t, processor.Process(context.Background(), message.RpcMessage{ID: 1, Body: tt.body}))
+ require.IsType(t, tt.wantType, sent)
+ if response, ok := sent.(*pb.BranchCommitResponseProto); ok {
+ require.Equal(t, pb.MessageTypeProto_TYPE_BRANCH_COMMIT_RESULT,
+ response.GetAbstractBranchEndResponse().GetAbstractTransactionResponse().GetAbstractResultMessage().GetAbstractMessage().GetMessageType())
+ }
+ })
+ }
}
diff --git a/pkg/remoting/processor/client/rm_branch_end_processor.go b/pkg/remoting/processor/client/rm_branch_end_processor.go
new file mode 100644
index 00000000..8478b4ea
--- /dev/null
+++ b/pkg/remoting/processor/client/rm_branch_end_processor.go
@@ -0,0 +1,80 @@
+/*
+ * 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 client
+
+import (
+ "errors"
+
+ "seata.apache.org/seata-go/v2/pkg/protocol/branch"
+ "seata.apache.org/seata-go/v2/pkg/protocol/message"
+ "seata.apache.org/seata-go/v2/pkg/remoting/grpc/pb"
+ "seata.apache.org/seata-go/v2/pkg/rm"
+)
+
+// branchEndResult is the protocol-independent result of a branch operation.
+type branchEndResult struct {
+ status branch.BranchStatus
+ resultCode message.ResultCode
+ errMsg string
+}
+
+func newBranchEndResult(status branch.BranchStatus, bizErr error) branchEndResult {
+ result := branchEndResult{status: status, resultCode: message.ResultCodeSuccess}
+ if bizErr != nil {
+ result.resultCode = message.ResultCodeFailed
+ result.errMsg = bizErr.Error()
+ }
+ return result
+}
+
+func branchEndResultCodeProto(resultCode message.ResultCode) pb.ResultCodeProto {
+ if resultCode == message.ResultCodeFailed {
+ return pb.ResultCodeProto_Failed
+ }
+ return pb.ResultCodeProto_Success
+}
+
+func branchEndSendResponse(
+ sendResponse func(int32, interface{}) error,
+ fallback func(int32, interface{}) error,
+) func(int32, interface{}) error {
+ if sendResponse != nil {
+ return sendResponse
+ }
+ return fallback
+}
+
+func branchEndProcessError(bizErr, sendErr error) error {
+ if sendErr == nil {
+ return nil
+ }
+ if bizErr == nil {
+ return sendErr
+ }
+ return errors.Join(bizErr, sendErr)
+}
+
+func branchEndResourceManager(
+ getResourceManager func(branch.BranchType) rm.ResourceManager,
+ branchType branch.BranchType,
+) rm.ResourceManager {
+ if getResourceManager != nil {
+ return getResourceManager(branchType)
+ }
+ return rm.GetRmCacheInstance().GetResourceManager(branchType)
+}
diff --git a/pkg/remoting/processor/client/rm_branch_rollback_processor.go b/pkg/remoting/processor/client/rm_branch_rollback_processor.go
index 40a7e86c..8f17d6f7 100644
--- a/pkg/remoting/processor/client/rm_branch_rollback_processor.go
+++ b/pkg/remoting/processor/client/rm_branch_rollback_processor.go
@@ -45,7 +45,10 @@ func initBranchRollback() {
}
}
-type rmBranchRollbackProcessor struct{}
+type rmBranchRollbackProcessor struct {
+ getResourceManager func(branch.BranchType) rm.ResourceManager
+ sendResponse func(int32, interface{}) error
+}
func (f *rmBranchRollbackProcessor) Process(ctx context.Context, rpcMessage message.RpcMessage) error {
log.Infof("the rm client received rmBranchRollback rpcMessage %#v from tc server.", rpcMessage)
@@ -73,45 +76,36 @@ func (f *rmBranchRollbackProcessor) handleGrpcBranchRollback(ctx context.Context
ResourceId: resourceID,
ApplicationData: []byte(applicationData),
}
- status, err := rm.GetRmCacheInstance().GetResourceManager(branchType).BranchRollback(ctx, branchResource)
- if err != nil {
- log.Errorf("branch rollback error: %s", err.Error())
- return err
- }
- log.Infof("branch rollback success: xid %s, branchID %d, resourceID %s, applicationData %s", xid, branchID, resourceID, applicationData)
-
- var (
- resultCode pb.ResultCodeProto
- errMsg string
- )
- if err != nil {
- resultCode = pb.ResultCodeProto_Failed
- errMsg = err.Error()
+ status, bizErr := branchEndResourceManager(f.getResourceManager, branchType).BranchRollback(ctx, branchResource)
+ if bizErr != nil {
+ log.Errorf("branch rollback error: xid %s, branchID %d: %s", xid, branchID, bizErr.Error())
} else {
- resultCode = pb.ResultCodeProto_Success
+ log.Infof("branch rollback success: xid %s, branchID %d, resourceID %s, applicationData %s", xid, branchID, resourceID, applicationData)
}
- // reply commit response to tc server
+ result := newBranchEndResult(status, bizErr)
+ // reply rollback response to tc server
response := &pb.BranchRollbackResponseProto{
AbstractBranchEndResponse: &pb.AbstractBranchEndResponseProto{
AbstractTransactionResponse: &pb.AbstractTransactionResponseProto{
AbstractResultMessage: &pb.AbstractResultMessageProto{
AbstractMessage: &pb.AbstractMessageProto{MessageType: pb.MessageTypeProto_TYPE_BRANCH_ROLLBACK_RESULT},
- ResultCode: resultCode,
- Msg: errMsg,
+ ResultCode: branchEndResultCodeProto(result.resultCode),
+ Msg: result.errMsg,
},
},
Xid: xid,
BranchId: branchID,
- BranchStatus: pb.BranchStatusProto(status),
+ BranchStatus: pb.BranchStatusProto(result.status),
},
}
- err = grpc.GetGrpcRemotingClient().SendAsyncResponse(rpcMessage.ID, response)
- if err != nil {
- log.Errorf("send branch rollback response error: {%#v}", err.Error())
- return err
+ sendResponse := branchEndSendResponse(f.sendResponse, grpc.GetGrpcRemotingClient().SendAsyncResponse)
+ sendErr := sendResponse(rpcMessage.ID, response)
+ if sendErr != nil {
+ log.Errorf("send branch rollback response error: xid %s, branchID %d: %v", xid, branchID, sendErr)
+ } else {
+ log.Infof("send branch rollback response success: xid %s, branchID %d, resourceID %s, applicationData %s", xid, branchID, resourceID, applicationData)
}
- log.Infof("send branch rollback response success: xid %s, branchID %d, resourceID %s, applicationData %s", xid, branchID, resourceID, applicationData)
- return nil
+ return branchEndProcessError(bizErr, sendErr)
}
func (f *rmBranchRollbackProcessor) handleGettyBranchRollback(ctx context.Context, rpcMessage message.RpcMessage) error {
@@ -128,42 +122,33 @@ func (f *rmBranchRollbackProcessor) handleGettyBranchRollback(ctx context.Contex
ResourceId: resourceID,
ApplicationData: applicationData,
}
- status, err := rm.GetRmCacheInstance().GetResourceManager(request.BranchType).BranchRollback(ctx, branchResource)
- if err != nil {
- log.Errorf("branch rollback error: %s", err.Error())
- return err
- }
- log.Infof("branch rollback success: xid %s, branchID %d, resourceID %s, applicationData %s", xid, branchID, resourceID, applicationData)
-
- var (
- resultCode message.ResultCode
- errMsg string
- )
- if err != nil {
- resultCode = message.ResultCodeFailed
- errMsg = err.Error()
+ status, bizErr := branchEndResourceManager(f.getResourceManager, request.BranchType).BranchRollback(ctx, branchResource)
+ if bizErr != nil {
+ log.Errorf("branch rollback error: xid %s, branchID %d: %s", xid, branchID, bizErr.Error())
} else {
- resultCode = message.ResultCodeSuccess
+ log.Infof("branch rollback success: xid %s, branchID %d, resourceID %s, applicationData %s", xid, branchID, resourceID, applicationData)
}
- // reply commit response to tc server
+ result := newBranchEndResult(status, bizErr)
+ // reply rollback response to tc server
response := message.BranchRollbackResponse{
AbstractBranchEndResponse: message.AbstractBranchEndResponse{
AbstractTransactionResponse: message.AbstractTransactionResponse{
AbstractResultMessage: message.AbstractResultMessage{
- ResultCode: resultCode,
- Msg: errMsg,
+ ResultCode: result.resultCode,
+ Msg: result.errMsg,
},
},
Xid: xid,
BranchId: branchID,
- BranchStatus: status,
+ BranchStatus: result.status,
},
}
- err = getty.GetGettyRemotingClient().SendAsyncResponse(rpcMessage.ID, response)
- if err != nil {
- log.Errorf("send branch rollback response error: {%#v}", err.Error())
- return err
+ sendResponse := branchEndSendResponse(f.sendResponse, getty.GetGettyRemotingClient().SendAsyncResponse)
+ sendErr := sendResponse(rpcMessage.ID, response)
+ if sendErr != nil {
+ log.Errorf("send branch rollback response error: xid %s, branchID %d: %v", xid, branchID, sendErr)
+ } else {
+ log.Infof("send branch rollback response success: xid %s, branchID %d, resourceID %s, applicationData %s", xid, branchID, resourceID, applicationData)
}
- log.Infof("send branch rollback response success: xid %s, branchID %d, resourceID %s, applicationData %s", xid, branchID, resourceID, applicationData)
- return nil
+ return branchEndProcessError(bizErr, sendErr)
}
diff --git a/pkg/remoting/processor/client/rm_branch_rollback_processor_test.go b/pkg/remoting/processor/client/rm_branch_rollback_processor_test.go
index 1683a786..3c1958cc 100644
--- a/pkg/remoting/processor/client/rm_branch_rollback_processor_test.go
+++ b/pkg/remoting/processor/client/rm_branch_rollback_processor_test.go
@@ -19,129 +19,177 @@ package client
import (
"context"
- "reflect"
+ "errors"
"testing"
- "github.com/agiledragon/gomonkey/v2"
- "github.com/stretchr/testify/assert"
-
- "seata.apache.org/seata-go/v2/pkg/rm/tcc"
-
- model2 "seata.apache.org/seata-go/v2/pkg/protocol/branch"
- "seata.apache.org/seata-go/v2/pkg/protocol/codec"
+ "github.com/stretchr/testify/require"
+ "seata.apache.org/seata-go/v2/pkg/protocol"
+ "seata.apache.org/seata-go/v2/pkg/protocol/branch"
"seata.apache.org/seata-go/v2/pkg/protocol/message"
"seata.apache.org/seata-go/v2/pkg/remoting/config"
- remotinggrpc "seata.apache.org/seata-go/v2/pkg/remoting/grpc"
"seata.apache.org/seata-go/v2/pkg/remoting/grpc/pb"
"seata.apache.org/seata-go/v2/pkg/rm"
)
-func TestRmBranchRollbackProcessor(t *testing.T) {
- // testcases
+func TestRmBranchRollbackProcessor_SendsFailureResponse(t *testing.T) {
+ bizErr := errors.New("rollback failed")
+ manager := &testResourceManager{rollbackStatus: branch.BranchStatusPhasetwoRollbackFailedRetryable, rollbackErr: bizErr}
+
+ t.Run("getty", func(t *testing.T) {
+ var sent interface{}
+ processor := rmBranchRollbackProcessor{
+ getResourceManager: func(branch.BranchType) rm.ResourceManager { return manager },
+ sendResponse: func(_ int32, response interface{}) error { sent = response; return nil },
+ }
+ err := processor.handleGettyBranchRollback(context.Background(), message.RpcMessage{ID: 1, Body: message.BranchRollbackRequest{
+ AbstractBranchEndRequest: message.AbstractBranchEndRequest{Xid: "xid", BranchId: 7, BranchType: branch.BranchTypeTCC, ResourceId: "resource"},
+ }})
+ require.NoError(t, err)
+ got := sent.(message.BranchRollbackResponse)
+ require.Equal(t, message.ResultCodeFailed, got.ResultCode)
+ require.Equal(t, bizErr.Error(), got.Msg)
+ require.Equal(t, manager.rollbackStatus, got.BranchStatus)
+ require.Equal(t, "xid", got.Xid)
+ require.Equal(t, int64(7), got.BranchId)
+ })
+
+ t.Run("grpc", func(t *testing.T) {
+ var sent interface{}
+ processor := rmBranchRollbackProcessor{
+ getResourceManager: func(branch.BranchType) rm.ResourceManager { return manager },
+ sendResponse: func(_ int32, response interface{}) error { sent = response; return nil },
+ }
+ err := processor.handleGrpcBranchRollback(context.Background(), message.RpcMessage{ID: 1, Body: &pb.BranchRollbackRequestProto{
+ AbstractBranchEndRequest: &pb.AbstractBranchEndRequestProto{Xid: "xid", BranchId: 7, BranchType: pb.BranchTypeProto_TCC, ResourceId: "resource"},
+ }})
+ require.NoError(t, err)
+ got := sent.(*pb.BranchRollbackResponseProto)
+ result := got.AbstractBranchEndResponse.AbstractTransactionResponse.AbstractResultMessage
+ require.Equal(t, pb.MessageTypeProto_TYPE_BRANCH_ROLLBACK_RESULT, result.GetAbstractMessage().GetMessageType())
+ require.Equal(t, pb.ResultCodeProto_Failed, result.ResultCode)
+ require.Equal(t, bizErr.Error(), result.Msg)
+ require.Equal(t, pb.BranchStatusProto(manager.rollbackStatus), got.AbstractBranchEndResponse.BranchStatus)
+ require.Equal(t, "xid", got.AbstractBranchEndResponse.Xid)
+ require.Equal(t, int64(7), got.AbstractBranchEndResponse.BranchId)
+ })
+}
+
+func TestRmBranchRollbackProcessor_ObservesBusinessAndSendErrors(t *testing.T) {
+ bizErr := errors.New("rollback failed")
+ sendErr := errors.New("send failed")
+ manager := &testResourceManager{rollbackStatus: branch.BranchStatusPhasetwoRollbackFailedRetryable, rollbackErr: bizErr}
+
+ t.Run("getty", func(t *testing.T) {
+ processor := rmBranchRollbackProcessor{
+ getResourceManager: func(branch.BranchType) rm.ResourceManager { return manager },
+ sendResponse: func(int32, interface{}) error { return sendErr },
+ }
+ err := processor.handleGettyBranchRollback(context.Background(), message.RpcMessage{ID: 1, Body: message.BranchRollbackRequest{
+ AbstractBranchEndRequest: message.AbstractBranchEndRequest{Xid: "xid", BranchId: 7, BranchType: branch.BranchTypeTCC, ResourceId: "resource"},
+ }})
+ require.ErrorIs(t, err, bizErr)
+ require.ErrorIs(t, err, sendErr)
+ })
+
+ t.Run("grpc", func(t *testing.T) {
+ processor := rmBranchRollbackProcessor{
+ getResourceManager: func(branch.BranchType) rm.ResourceManager { return manager },
+ sendResponse: func(int32, interface{}) error { return sendErr },
+ }
+ err := processor.handleGrpcBranchRollback(context.Background(), message.RpcMessage{ID: 1, Body: &pb.BranchRollbackRequestProto{
+ AbstractBranchEndRequest: &pb.AbstractBranchEndRequestProto{Xid: "xid", BranchId: 7, BranchType: pb.BranchTypeProto_TCC, ResourceId: "resource"},
+ }})
+ require.ErrorIs(t, err, bizErr)
+ require.ErrorIs(t, err, sendErr)
+ })
+}
+
+func TestRmBranchRollbackProcessor_SendsSuccessResponse(t *testing.T) {
+ manager := &testResourceManager{rollbackStatus: branch.BranchStatusPhasetwoRollbacked}
+
+ t.Run("getty", func(t *testing.T) {
+ var sent interface{}
+ processor := rmBranchRollbackProcessor{
+ getResourceManager: func(branch.BranchType) rm.ResourceManager { return manager },
+ sendResponse: func(_ int32, response interface{}) error { sent = response; return nil },
+ }
+ err := processor.handleGettyBranchRollback(context.Background(), message.RpcMessage{ID: 1, Body: message.BranchRollbackRequest{
+ AbstractBranchEndRequest: message.AbstractBranchEndRequest{Xid: "xid", BranchId: 7, BranchType: branch.BranchTypeTCC, ResourceId: "resource"},
+ }})
+ require.NoError(t, err)
+ got := sent.(message.BranchRollbackResponse)
+ require.Equal(t, message.ResultCodeSuccess, got.ResultCode)
+ require.Empty(t, got.Msg)
+ require.Equal(t, manager.rollbackStatus, got.BranchStatus)
+ require.Equal(t, "xid", got.Xid)
+ require.Equal(t, int64(7), got.BranchId)
+ })
+
+ t.Run("grpc", func(t *testing.T) {
+ var sent interface{}
+ processor := rmBranchRollbackProcessor{
+ getResourceManager: func(branch.BranchType) rm.ResourceManager { return manager },
+ sendResponse: func(_ int32, response interface{}) error { sent = response; return nil },
+ }
+ err := processor.handleGrpcBranchRollback(context.Background(), message.RpcMessage{ID: 1, Body: &pb.BranchRollbackRequestProto{
+ AbstractBranchEndRequest: &pb.AbstractBranchEndRequestProto{Xid: "xid", BranchId: 7, BranchType: pb.BranchTypeProto_TCC, ResourceId: "resource"},
+ }})
+ require.NoError(t, err)
+ got := sent.(*pb.BranchRollbackResponseProto)
+ result := got.AbstractBranchEndResponse.AbstractTransactionResponse.AbstractResultMessage
+ require.Equal(t, pb.MessageTypeProto_TYPE_BRANCH_ROLLBACK_RESULT, result.GetAbstractMessage().GetMessageType())
+ require.Equal(t, pb.ResultCodeProto_Success, result.ResultCode)
+ require.Empty(t, result.Msg)
+ require.Equal(t, pb.BranchStatusProto(manager.rollbackStatus), got.AbstractBranchEndResponse.BranchStatus)
+ require.Equal(t, "xid", got.AbstractBranchEndResponse.Xid)
+ require.Equal(t, int64(7), got.AbstractBranchEndResponse.BranchId)
+ })
+}
+
+func TestRmBranchRollbackProcessor_ProcessRoutesByProtocol(t *testing.T) {
+ previous := config.GetTransportConfig()
+ t.Cleanup(func() { config.InitTransportConfig(previous) })
+
+ manager := &testResourceManager{rollbackStatus: branch.BranchStatusPhasetwoRollbacked}
tests := []struct {
- name string // testcase name
- protocol string // protocol:seata/grpc
- rpcMsg message.RpcMessage // rpcMessage case
- wantErr bool // want testcase err or not
+ name string
+ protocol protocol.Protocol
+ body interface{}
+ wantType interface{}
}{
{
- name: "rbr-testcase1-failure",
- protocol: "seata",
- rpcMsg: message.RpcMessage{
- ID: 223,
- Type: message.RequestType(message.MessageTypeBranchRollback),
- Codec: byte(codec.CodecTypeSeata),
- Compressor: byte(1),
- HeadMap: map[string]string{
- "name": " Jack",
- "age": "12",
- "address": "Beijing",
- },
- Body: message.BranchRollbackRequest{
- AbstractBranchEndRequest: message.AbstractBranchEndRequest{
- Xid: "123345",
- BranchId: 56679,
- BranchType: model2.BranchTypeTCC,
- ResourceId: "1232324",
- ApplicationData: []byte("TestExtraData"),
- },
- },
- },
-
- wantErr: true, // need dail to server, so err accured
+ name: "seata",
+ protocol: protocol.ProtocolSEATA,
+ body: message.BranchRollbackRequest{AbstractBranchEndRequest: message.AbstractBranchEndRequest{
+ Xid: "xid", BranchId: 7, BranchType: branch.BranchTypeTCC, ResourceId: "resource",
+ }},
+ wantType: message.BranchRollbackResponse{},
},
{
- name: "rbr-testcase2-failure",
- protocol: "grpc",
- rpcMsg: message.RpcMessage{
- ID: 223,
- Type: message.RequestType(message.MessageTypeBranchRollback),
- HeadMap: map[string]string{
- "name": " Jack",
- "age": "12",
- "address": "Beijing",
- },
- Body: &pb.BranchRollbackRequestProto{
- AbstractBranchEndRequest: &pb.AbstractBranchEndRequestProto{
- Xid: "123345",
- BranchId: 56679,
- BranchType: pb.BranchTypeProto_TCC,
- ResourceId: "1232324",
- ApplicationData: "TestExtraData",
- },
- },
- },
-
- wantErr: true, // need dail to server, so err accured
+ name: "grpc",
+ protocol: protocol.ProtocolGRPC,
+ body: &pb.BranchRollbackRequestProto{AbstractBranchEndRequest: &pb.AbstractBranchEndRequestProto{
+ Xid: "xid", BranchId: 7, BranchType: pb.BranchTypeProto_TCC, ResourceId: "resource",
+ }},
+ wantType: (*pb.BranchRollbackResponseProto)(nil),
},
}
- var ctx context.Context
- var rbrProcessor rmBranchRollbackProcessor
-
- // run tests
- for _, tc := range tests {
- t.Run(tc.name, func(t *testing.T) {
- config.InitTransportConfig(&config.TransportConfig{Protocol: tc.protocol})
-
- rm.GetRmCacheInstance().RegisterResourceManager(tcc.GetTCCResourceManagerInstance())
-
- err := rbrProcessor.Process(ctx, tc.rpcMsg)
- if (err != nil) != tc.wantErr {
- t.Errorf("rmBranchRollbackProcessor wantErr: %v, got: %v", tc.wantErr, err)
- return
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ config.InitTransportConfig(&config.TransportConfig{Protocol: tt.protocol.String()})
+ var sent interface{}
+ processor := rmBranchRollbackProcessor{
+ getResourceManager: func(branch.BranchType) rm.ResourceManager { return manager },
+ sendResponse: func(_ int32, response interface{}) error { sent = response; return nil },
}
- })
- }
-}
-func TestRmBranchRollbackProcessorResponseUsesRollbackResultMessageType(t *testing.T) {
- resourceManager := &testGrpcResourceManager{branchType: model2.BranchTypeTCC}
- rm.GetRmCacheInstance().RegisterResourceManager(resourceManager)
- defer rm.GetRmCacheInstance().UnregisterResourceManager(model2.BranchTypeTCC)
- config.InitTransportConfig(&config.TransportConfig{Protocol: "grpc"})
-
- var response *pb.BranchRollbackResponseProto
- patches := gomonkey.ApplyMethod(reflect.TypeOf(remotinggrpc.GetGrpcRemotingClient()), "SendAsyncResponse",
- func(_ *remotinggrpc.GrpcRemotingClient, _ int32, msg interface{}) error {
- response = msg.(*pb.BranchRollbackResponseProto)
- return nil
+ require.NoError(t, processor.Process(context.Background(), message.RpcMessage{ID: 1, Body: tt.body}))
+ require.IsType(t, tt.wantType, sent)
+ if response, ok := sent.(*pb.BranchRollbackResponseProto); ok {
+ require.Equal(t, pb.MessageTypeProto_TYPE_BRANCH_ROLLBACK_RESULT,
+ response.GetAbstractBranchEndResponse().GetAbstractTransactionResponse().GetAbstractResultMessage().GetAbstractMessage().GetMessageType())
+ }
})
- defer patches.Reset()
-
- err := (&rmBranchRollbackProcessor{}).Process(context.Background(), message.RpcMessage{
- ID: 1,
- Type: message.RequestType(message.MessageTypeBranchRollback),
- Body: &pb.BranchRollbackRequestProto{AbstractBranchEndRequest: &pb.AbstractBranchEndRequestProto{
- Xid: "test-xid",
- BranchId: 1,
- BranchType: pb.BranchTypeProto_TCC,
- }},
- })
-
- assert.NoError(t, err)
- if assert.NotNil(t, response) {
- assert.Equal(t, pb.MessageTypeProto_TYPE_BRANCH_ROLLBACK_RESULT,
- response.GetAbstractBranchEndResponse().GetAbstractTransactionResponse().GetAbstractResultMessage().GetAbstractMessage().GetMessageType())
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org