fix xa branch commit bug (apache#564)

luky116 · web-flow · commit bfb9eb12f4ff · 2023-05-28T15:43:48.000+08:00
* fix xa branch commit bug
diff --git a/goimports.sh b/goimports.sh
@@ -16,7 +16,7 @@
 #
 
 # format go imports style
-go install -v golang.org/x/tools/cmd/goimports
+go install golang.org/x/tools/cmd/goimports
 goimports  -local github.com/seata/seata-go -w .
 
 # format licence style
diff --git a/pkg/client/client.go b/pkg/client/client.go
@@ -18,14 +18,14 @@
 package client
 
 import (
-	"github.com/seata/seata-go/pkg/remoting/getty"
 	"sync"
 
 	"github.com/seata/seata-go/pkg/datasource"
 	at "github.com/seata/seata-go/pkg/datasource/sql"
 	"github.com/seata/seata-go/pkg/datasource/sql/exec/config"
 	"github.com/seata/seata-go/pkg/integration"
 	remoteConfig "github.com/seata/seata-go/pkg/remoting/config"
+	"github.com/seata/seata-go/pkg/remoting/getty"
 	"github.com/seata/seata-go/pkg/remoting/processor/client"
 	"github.com/seata/seata-go/pkg/rm"
 	"github.com/seata/seata-go/pkg/rm/tcc"
diff --git a/pkg/datasource/sql/conn_xa.go b/pkg/datasource/sql/conn_xa.go
@@ -322,22 +322,22 @@ func (c *XAConn) Commit(ctx context.Context) error {
 
 	now := time.Now()
 	if c.end(ctx, xa.TMSuccess) != nil {
-		return c.commitErrorHandle()
+		return c.commitErrorHandle(ctx)
 	}
 
 	if c.checkTimeout(ctx, now) != nil {
-		return c.commitErrorHandle()
+		return c.commitErrorHandle(ctx)
 	}
 
 	if c.xaResource.XAPrepare(ctx, c.xaBranchXid.String()) != nil {
-		return c.commitErrorHandle()
+		return c.commitErrorHandle(ctx)
 	}
 	return nil
 }
 
-func (c *XAConn) commitErrorHandle() error {
+func (c *XAConn) commitErrorHandle(ctx context.Context) error {
 	var err error
-	if err = c.tx.Rollback(); err != nil {
+	if err = c.XaRollback(ctx, c.xaBranchXid); err != nil {
 		err = fmt.Errorf("failed to report XA branch commit-failure xid:%s, err:%w", c.txCtx.XID, err)
 	}
 	c.cleanXABranchContext()
@@ -389,7 +389,7 @@ func (c *XAConn) XaRollbackByBranchId(ctx context.Context, xaXid XAXid) error {
 }
 
 func (c *XAConn) XaRollback(ctx context.Context, xaXid XAXid) error {
-	err := c.xaResource.Rollback(ctx, xaXid.GetGlobalXid())
+	err := c.xaResource.Rollback(ctx, xaXid.String())
 	c.releaseIfNecessary()
 	return err
 }
diff --git a/pkg/datasource/sql/db.go b/pkg/datasource/sql/db.go
@@ -22,14 +22,15 @@ import (
 	"database/sql"
 	"database/sql/driver"
 	"fmt"
-	"github.com/seata/seata-go/pkg/util/log"
 	"sync"
 
 	"github.com/seata/seata-go/pkg/datasource/sql/datasource"
 	"github.com/seata/seata-go/pkg/datasource/sql/types"
 	"github.com/seata/seata-go/pkg/datasource/sql/undo"
 	"github.com/seata/seata-go/pkg/datasource/sql/util"
+	"github.com/seata/seata-go/pkg/datasource/sql/xa"
 	"github.com/seata/seata-go/pkg/protocol/branch"
+	"github.com/seata/seata-go/pkg/util/log"
 )
 
 type dbOption func(db *DBResource)
@@ -218,10 +219,17 @@ func (db *DBResource) ConnectionForXA(ctx context.Context, xaXid XAXid) (*XAConn
 	if err != nil {
 		return nil, fmt.Errorf("get xa new connection failure, xid:%s, err:%v", xaXid.String(), err)
 	}
+	xaResource, err := xa.CreateXAResource(newDriverConn, types.DBTypeMySQL)
+	if err != nil {
+		return nil, fmt.Errorf("create xa resoruce err:%w", err)
+	}
 	xaConn := &XAConn{
 		Conn: &Conn{
 			targetConn: newDriverConn,
+			res:        db,
 		},
+		xaBranchXid: XaIdBuild(xaXid.GetGlobalXid(), xaXid.GetBranchId()),
+		xaResource:  xaResource,
 	}
 	return xaConn, nil
 }
diff --git a/pkg/datasource/sql/driver.go b/pkg/datasource/sql/driver.go
@@ -28,6 +28,7 @@ import (
 	"strings"
 
 	"github.com/go-sql-driver/mysql"
+
 	"github.com/seata/seata-go/pkg/datasource/sql/datasource"
 	mysql2 "github.com/seata/seata-go/pkg/datasource/sql/datasource/mysql"
 	"github.com/seata/seata-go/pkg/datasource/sql/types"
diff --git a/pkg/datasource/sql/exec/at/multi_delete_executor.go b/pkg/datasource/sql/exec/at/multi_delete_executor.go
@@ -70,7 +70,7 @@ type multiDelete struct {
 	clear bool
 }
 
-//NewMultiDeleteExecutor get multiDelete executor
+// NewMultiDeleteExecutor get multiDelete executor
 func NewMultiDeleteExecutor(parserCtx *types.ParseContext, execContent *types.ExecContext, hooks []exec.SQLHook) *multiDeleteExecutor {
 	return &multiDeleteExecutor{parserCtx: parserCtx, execContext: execContent, baseExecutor: baseExecutor{hooks: hooks}}
 }
diff --git a/pkg/datasource/sql/xa/mysql_xa_connection.go b/pkg/datasource/sql/xa/mysql_xa_connection.go
@@ -25,6 +25,8 @@ import (
 	"io"
 	"strings"
 	"time"
+
+	"github.com/seata/seata-go/pkg/util/log"
 )
 
 type MysqlXAConn struct {
@@ -36,6 +38,8 @@ func NewMysqlXaConn(conn driver.Conn) *MysqlXAConn {
 }
 
 func (c *MysqlXAConn) Commit(ctx context.Context, xid string, onePhase bool) error {
+	log.Infof("xa branch commit, xid %s", xid)
+
 	var sb strings.Builder
 	sb.WriteString("XA COMMIT ")
 	sb.WriteString("'")
@@ -47,10 +51,15 @@ func (c *MysqlXAConn) Commit(ctx context.Context, xid string, onePhase bool) err
 
 	conn, _ := c.Conn.(driver.ExecerContext)
 	_, err := conn.ExecContext(ctx, sb.String(), nil)
+	if err != nil {
+		log.Errorf("xa branch commit failed, xid %s, err %v", xid, err)
+	}
 	return err
 }
 
 func (c *MysqlXAConn) End(ctx context.Context, xid string, flags int) error {
+	log.Infof("xa branch end, xid %s", xid)
+
 	var sb strings.Builder
 	sb.WriteString("XA END ")
 	sb.WriteString("'")
@@ -71,6 +80,9 @@ func (c *MysqlXAConn) End(ctx context.Context, xid string, flags int) error {
 
 	conn, _ := c.Conn.(driver.ExecerContext)
 	_, err := conn.ExecContext(ctx, sb.String(), nil)
+	if err != nil {
+		log.Errorf("xa branch end failed, xid %s, err %v", xid, err)
+	}
 	return err
 }
 
@@ -91,6 +103,8 @@ func (c *MysqlXAConn) IsSameRM(ctx context.Context, xares XAResource) bool {
 }
 
 func (c *MysqlXAConn) XAPrepare(ctx context.Context, xid string) error {
+	log.Infof("xa branch prepare, xid %s", xid)
+
 	var sb strings.Builder
 	sb.WriteString("XA PREPARE ")
 	sb.WriteString("'")
@@ -99,6 +113,9 @@ func (c *MysqlXAConn) XAPrepare(ctx context.Context, xid string) error {
 
 	conn, _ := c.Conn.(driver.ExecerContext)
 	_, err := conn.ExecContext(ctx, sb.String(), nil)
+	if err != nil {
+		log.Errorf("xa branch prepare failed, xid %s, err %v", xid, err)
+	}
 	return err
 }
 
@@ -143,6 +160,8 @@ func (c *MysqlXAConn) Recover(ctx context.Context, flag int) (xids []string, err
 }
 
 func (c *MysqlXAConn) Rollback(ctx context.Context, xid string) error {
+	log.Infof("xa branch rollback, xid %s", xid)
+
 	var sb strings.Builder
 	sb.WriteString("XA ROLLBACK ")
 	sb.WriteString("'")
@@ -151,6 +170,9 @@ func (c *MysqlXAConn) Rollback(ctx context.Context, xid string) error {
 
 	conn, _ := c.Conn.(driver.ExecerContext)
 	_, err := conn.ExecContext(ctx, sb.String(), nil)
+	if err != nil {
+		log.Errorf("xa branch rollback failed, xid %s, err %v", xid, err)
+	}
 	return err
 }
 
@@ -159,6 +181,8 @@ func (c *MysqlXAConn) SetTransactionTimeout(duration time.Duration) bool {
 }
 
 func (c *MysqlXAConn) Start(ctx context.Context, xid string, flags int) error {
+	log.Infof("xa branch start, xid %s", xid)
+
 	var sb strings.Builder
 	sb.WriteString("XA START ")
 	sb.WriteString("'")
@@ -180,5 +204,8 @@ func (c *MysqlXAConn) Start(ctx context.Context, xid string, flags int) error {
 
 	conn, _ := c.Conn.(driver.ExecerContext)
 	_, err := conn.ExecContext(ctx, sb.String(), nil)
+	if err != nil {
+		log.Errorf("xa branch start failed, xid %s, err %v", xid, err)
+	}
 	return err
 }

Original file line number	Diff line number	Diff line change
`@@ -16,7 +16,7 @@`
`16`	`16`	`#`
`17`	`17`
`18`	`18`	`# format go imports style`
`19`		`-go install -v golang.org/x/tools/cmd/goimports`
	`19`	`+go install golang.org/x/tools/cmd/goimports`
`20`	`20`	`goimports -local github.com/seata/seata-go -w .`
`21`	`21`
`22`	`22`	`# format licence style`
Original file line number	Diff line number	Diff line change
`@@ -322,22 +322,22 @@ func (c *XAConn) Commit(ctx context.Context) error {`
`322`	`322`
`323`	`323`	`now := time.Now()`
`324`	`324`	`if c.end(ctx, xa.TMSuccess) != nil {`
`325`		`- return c.commitErrorHandle()`
	`325`	`+ return c.commitErrorHandle(ctx)`
`326`	`326`	`}`
`327`	`327`
`328`	`328`	`if c.checkTimeout(ctx, now) != nil {`
`329`		`- return c.commitErrorHandle()`
	`329`	`+ return c.commitErrorHandle(ctx)`
`330`	`330`	`}`
`331`	`331`
`332`	`332`	`if c.xaResource.XAPrepare(ctx, c.xaBranchXid.String()) != nil {`
`333`		`- return c.commitErrorHandle()`
	`333`	`+ return c.commitErrorHandle(ctx)`
`334`	`334`	`}`
`335`	`335`	`return nil`
`336`	`336`	`}`
`337`	`337`
`338`		`-func (c *XAConn) commitErrorHandle() error {`
	`338`	`+func (c *XAConn) commitErrorHandle(ctx context.Context) error {`
`339`	`339`	`var err error`
`340`		`- if err = c.tx.Rollback(); err != nil {`
	`340`	`+ if err = c.XaRollback(ctx, c.xaBranchXid); err != nil {`
`341`	`341`	`err = fmt.Errorf("failed to report XA branch commit-failure xid:%s, err:%w", c.txCtx.XID, err)`
`342`	`342`	`}`
`343`	`343`	`c.cleanXABranchContext()`
`@@ -389,7 +389,7 @@ func (c *XAConn) XaRollbackByBranchId(ctx context.Context, xaXid XAXid) error {`
`389`	`389`	`}`
`390`	`390`
`391`	`391`	`func (c *XAConn) XaRollback(ctx context.Context, xaXid XAXid) error {`
`392`		`- err := c.xaResource.Rollback(ctx, xaXid.GetGlobalXid())`
	`392`	`+ err := c.xaResource.Rollback(ctx, xaXid.String())`
`393`	`393`	`c.releaseIfNecessary()`
`394`	`394`	`return err`
`395`	`395`	`}`
Original file line number	Diff line number	Diff line change
`@@ -70,7 +70,7 @@ type multiDelete struct {`
`70`	`70`	`clear bool`
`71`	`71`	`}`
`72`	`72`
`73`		`-//NewMultiDeleteExecutor get multiDelete executor`
	`73`	`+// NewMultiDeleteExecutor get multiDelete executor`
`74`	`74`	`func NewMultiDeleteExecutor(parserCtx types.ParseContext, execContent types.ExecContext, hooks []exec.SQLHook) *multiDeleteExecutor {`
`75`	`75`	`return &multiDeleteExecutor{parserCtx: parserCtx, execContext: execContent, baseExecutor: baseExecutor{hooks: hooks}}`
`76`	`76`	`}`