Repository: apache/pulsar-client-go
Pull request #1318 (2024-12-12 00:01:16)

chore(deps): bump golang.org/x/crypto from 0.22.0 to 0.31.0
Bumps [golang.org/x/crypto](https://github.com/golang/crypto) from 0.22.0 to 0.31.0.
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/golang/crypto/commit/b4f1988a35dee11ec3e05d6bf3e90b695fbd8909"><code>b4f1988</code></a> ssh: make the public key cache a 1-entry FIFO cache</li>
<li><a href="https://github.com/golang/crypto/commit/7042ebcbe097f305ba3a93f9a22b4befa4b83d29"><code>7042ebc</code></a> openpgp/clearsign: just use rand.Reader in tests</li>
<li><a href="https://github.com/golang/crypto/commit/3e90321ac7bcee3d924ed63ed3ad97be2079cb56"><code>3e90321</code></a> go.mod: update golang.org/x dependencies</li>
<li><a href="https://github.com/golang/crypto/commit/8c4e668694ccbaa1be4785da7e7a40f2ef93152b"><code>8c4e668</code></a> x509roots/fallback: update bundle</li>
<li><a href="https://github.com/golang/crypto/commit/6018723c74059e3b91c84268b212c2f6cdab1f64"><code>6018723</code></a> go.mod: update golang.org/x dependencies</li>
<li><a href="https://github.com/golang/crypto/commit/71ed71b4faf97caafd1863fed003e9ac311f10ee"><code>71ed71b</code></a> README: don't recommend go get</li>
<li><a href="https://github.com/golang/crypto/commit/750a45fe5e473d5afa193e9088f3d135e64eca26"><code>750a45f</code></a> sha3: add MarshalBinary, AppendBinary, and UnmarshalBinary</li>
<li><a href="https://github.com/golang/crypto/commit/36b172546bd03a74c79e109ec84c599b672ea9e4"><code>36b1725</code></a> sha3: avoid trailing permutation</li>
<li><a href="https://github.com/golang/crypto/commit/80ea76eb17c0c52f5d5d04e833d6aeb6b062d81d"><code>80ea76e</code></a> sha3: fix padding for long cSHAKE parameters</li>
<li><a href="https://github.com/golang/crypto/commit/c17aa50fbd32393e5d52fa65ca51cbfff0a75aea"><code>c17aa50</code></a> sha3: avoid buffer copy</li>
<li>Additional commits viewable in <a href="https://github.com/golang/crypto/compare/v0.22.0...v0.31.0">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=golang.org/x/crypto&package-manager=go_modules&previous-version=0.22.0&new-version=0.31.0)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot show <dependency name> ignore conditions` will show all of the ignore conditions of the specified dependency
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)
You can disable automated security fix PRs for this repo from the [Security Alerts page](https://github.com/apache/pulsar-client-go/network/alerts).

</details>

nodece: @dependabot rebase

Patch:
diff --git a/.golangci.yml b/.golangci.yml
index 1f1b42c..58ecd75 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -18,7 +18,7 @@
 # Run `make lint` from the root path of this project to check code with golangci-lint.
 
 run:
-  deadline: 6m
+  timeout: 5m
 
 linters:
   # Uncomment this line to run only the explicitly enabled linters
diff --git a/go.mod b/go.mod
index c2f3457..088c4b4 100644
--- a/go.mod
+++ b/go.mod
@@ -96,11 +96,11 @@ require (
 	go.opentelemetry.io/otel v1.24.0 // indirect
 	go.opentelemetry.io/otel/metric v1.24.0 // indirect
 	go.opentelemetry.io/otel/trace v1.24.0 // indirect
-	golang.org/x/crypto v0.22.0 // indirect
+	golang.org/x/crypto v0.31.0 // indirect
 	golang.org/x/net v0.23.0 // indirect
-	golang.org/x/sys v0.19.0 // indirect
-	golang.org/x/term v0.19.0 // indirect
-	golang.org/x/text v0.14.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/term v0.27.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20231016165738-49dd2c1f3d0b // indirect
 	google.golang.org/grpc v1.59.0 // indirect
diff --git a/go.sum b/go.sum
index b546ffa..2d6ae68 100644
--- a/go.sum
+++ b/go.sum
@@ -419,8 +419,8 @@ golang.org/x/crypto v0.0.0-20190605123033-f99c8df09eb5/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.22.0 h1:g1v0xeRhjcugydODzvb3mEM9SQ0HGp9s/nh3COQ/C30=
-golang.org/x/crypto v0.22.0/go.mod h1:vr6Su+7cTlO45qkww3VDJlzDn0ctJvRgYbC2NvXHt+M=
+golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
+golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8=
@@ -562,12 +562,12 @@ golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/sys v0.19.0 h1:q5f1RH2jigJ1MoAWp2KTp3gm5zAGFUTarQZ5U386+4o=
-golang.org/x/sys v0.19.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
+golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
-golang.org/x/term v0.19.0 h1:+ThwsDv+tYfnJFhF4L8jITxu1tdTWRTZpdsWgEgjL6Q=
-golang.org/x/term v0.19.0/go.mod h1:2CuTdWZ7KHSQwUzKva0cbMg6q2DMI3Mmxp+gKJbskEk=
+golang.org/x/term v0.27.0 h1:WP60Sv1nlK1T6SupCHbXzSaN0b9wUmsPoRS9b61A23Q=
+golang.org/x/term v0.27.0/go.mod h1:iMsnZpn0cago0GOrHO2+Y7u7JPn5AylBrcoWkElMTSM=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -575,8 +575,8 @@ golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
-golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
-golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
+golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
+golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index 568dcaa..98848d7 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -532,8 +532,8 @@ func (pc *partitionConsumer) internalAckWithTxn(req *ackWithTxnRequest) {
 		req.err = newError(ConsumerClosed, "Failed to ack by closing or closed consumer")
 		return
 	}
-	if req.Transaction.state != TxnOpen {
-		pc.log.WithField("state", req.Transaction.state).Error("Failed to ack by a non-open transaction.")
+	if req.Transaction.state.Load() != int32(TxnOpen) {
+		pc.log.WithField("state", req.Transaction.state.Load()).Error("Failed to ack by a non-open transaction.")
 		req.err = newError(InvalidStatus, "Failed to ack by a non-open transaction.")
 		return
 	}
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index 7694c96..371ffb6 100755
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -1135,8 +1135,8 @@ func (p *partitionProducer) prepareTransaction(sr *sendRequest) error {
 	}
 
 	txn := (sr.msg.Transaction).(*transaction)
-	if txn.state != TxnOpen {
-		p.log.WithField("state", txn.state).Error("Failed to send message" +
+	if txn.state.Load() != int32(TxnOpen) {
+		p.log.WithField("state", txn.state.Load()).Error("Failed to send message" +
 			" by a non-open transaction.")
 		return joinErrors(ErrTransaction,
 			fmt.Errorf("failed to send message by a non-open transaction"))
diff --git a/pulsar/transaction_impl.go b/pulsar/transaction_impl.go
index 82e6fcf..a489781 100644
--- a/pulsar/transaction_impl.go
+++ b/pulsar/transaction_impl.go
@@ -19,6 +19,8 @@ package pulsar
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -33,9 +35,9 @@ type subscription struct {
 }
 
 type transaction struct {
-	sync.Mutex
+	mu                       sync.Mutex
 	txnID                    TxnID
-	state                    TxnState
+	state                    atomic.Int32
 	tcClient                 *transactionCoordinatorClient
 	registerPartitions       map[string]bool
 	registerAckSubscriptions map[subscription]bool
@@ -54,7 +56,7 @@ type transaction struct {
 	//     1. When the transaction is committed or aborted, a bool will be read from opsFlow chan.
 	//     2. When the opsCount increment from 0 to 1, a bool will be read from opsFlow chan.
 	opsFlow   chan bool
-	opsCount  int32
+	opsCount  atomic.Int32
 	opTimeout time.Duration
 	log       log.Logger
 }
@@ -62,47 +64,52 @@ type transaction struct {
 func newTransaction(id TxnID, tcClient *transactionCoordinatorClient, timeout time.Duration) *transaction {
 	transaction := &transaction{
 		txnID:                    id,
-		state:                    TxnOpen,
 		registerPartitions:       make(map[string]bool),
 		registerAckSubscriptions: make(map[subscription]bool),
 		opsFlow:                  make(chan bool, 1),
-		opTimeout:                5 * time.Second,
+		opTimeout:                tcClient.client.operationTimeout,
 		tcClient:                 tcClient,
 	}
-	//This means there are not pending requests with this transaction. The transaction can be committed or aborted.
+	transaction.state.Store(int32(TxnOpen))
+	// This means there are not pending requests with this transaction. The transaction can be committed or aborted.
 	transaction.opsFlow <- true
 	go func() {
-		//Set the state of the transaction to timeout after timeout
+		// Set the state of the transaction to timeout after timeout
 		<-time.After(timeout)
-		atomic.CompareAndSwapInt32((*int32)(&transaction.state), int32(TxnOpen), int32(TxnTimeout))
+		transaction.state.CompareAndSwap(int32(TxnOpen), int32(TxnTimeout))
 	}()
 	transaction.log = tcClient.log.SubLogger(log.Fields{})
 	return transaction
 }
 
 func (txn *transaction) GetState() TxnState {
-	return txn.state
+	return TxnState(txn.state.Load())
 }
 
-func (txn *transaction) Commit(_ context.Context) error {
-	if !(atomic.CompareAndSwapInt32((*int32)(&txn.state), int32(TxnOpen), int32(TxnCommitting)) ||
-		txn.state == TxnCommitting) {
-		return newError(InvalidStatus, "Expect transaction state is TxnOpen but "+txn.state.string())
+func (txn *transaction) Commit(ctx context.Context) error {
+	if !(txn.state.CompareAndSwap(int32(TxnOpen), int32(TxnCommitting))) {
+		txnState := txn.state.Load()
+		return newError(InvalidStatus, txnStateErrorMessage(TxnOpen, TxnState(txnState)))
 	}
 
-	//Wait for all operations to complete
+	// Wait for all operations to complete
 	select {
 	case <-txn.opsFlow:
+	case <-ctx.Done():
+		txn.state.Store(int32(TxnOpen))
+		return ctx.Err()
 	case <-time.After(txn.opTimeout):
+		txn.state.Store(int32(TxnTimeout))
 		return newError(TimeoutError, "There are some operations that are not completed after the timeout.")
 	}
-	//Send commit transaction command to transaction coordinator
+	// Send commit transaction command to transaction coordinator
 	err := txn.tcClient.endTxn(&txn.txnID, pb.TxnAction_COMMIT)
 	if err == nil {
-		atomic.StoreInt32((*int32)(&txn.state), int32(TxnCommitted))
+		txn.state.Store(int32(TxnCommitted))
 	} else {
-		if e, ok := err.(*Error); ok && (e.Result() == TransactionNoFoundError || e.Result() == InvalidStatus) {
-			atomic.StoreInt32((*int32)(&txn.state), int32(TxnError))
+		var e *Error
+		if errors.As(err, &e) && (e.Result() == TransactionNoFoundError || e.Result() == InvalidStatus) {
+			txn.state.Store(int32(TxnError))
 			return err
 		}
 		txn.opsFlow <- true
@@ -110,40 +117,45 @@ func (txn *transaction) Commit(_ context.Context) error {
 	return err
 }
 
-func (txn *transaction) Abort(_ context.Context) error {
-	if !(atomic.CompareAndSwapInt32((*int32)(&txn.state), int32(TxnOpen), int32(TxnAborting)) ||
-		txn.state == TxnAborting) {
-		return newError(InvalidStatus, "Expect transaction state is TxnOpen but "+txn.state.string())
+func (txn *transaction) Abort(ctx context.Context) error {
+	if !(txn.state.CompareAndSwap(int32(TxnOpen), int32(TxnAborting))) {
+		txnState := txn.state.Load()
+		return newError(InvalidStatus, txnStateErrorMessage(TxnOpen, TxnState(txnState)))
 	}
 
-	//Wait for all operations to complete
+	// Wait for all operations to complete
 	select {
 	case <-txn.opsFlow:
+	case <-ctx.Done():
+		txn.state.Store(int32(TxnOpen))
+		return ctx.Err()
 	case <-time.After(txn.opTimeout):
+		txn.state.Store(int32(TxnTimeout))
 		return newError(TimeoutError, "There are some operations that are not completed after the timeout.")
 	}
-	//Send abort transaction command to transaction coordinator
+	// Send abort transaction command to transaction coordinator
 	err := txn.tcClient.endTxn(&txn.txnID, pb.TxnAction_ABORT)
 	if err == nil {
-		atomic.StoreInt32((*int32)(&txn.state), int32(TxnAborted))
+		txn.state.Store(int32(TxnAborted))
 	} else {
-		if e, ok := err.(*Error); ok && (e.Result() == TransactionNoFoundError || e.Result() == InvalidStatus) {
-			atomic.StoreInt32((*int32)(&txn.state), int32(TxnError))
-		} else {
-			txn.opsFlow <- true
+		var e *Error
+		if errors.As(err, &e) && (e.Result() == TransactionNoFoundError || e.Result() == InvalidStatus) {
+			txn.state.Store(int32(TxnError))
+			return err
 		}
+		txn.opsFlow <- true
 	}
 	return err
 }
 
 func (txn *transaction) registerSendOrAckOp() error {
-	if atomic.AddInt32(&txn.opsCount, 1) == 1 {
-		//There are new operations that not completed
+	if txn.opsCount.Add(1) == 1 {
+		// There are new operations that were not completed
 		select {
 		case <-txn.opsFlow:
 			return nil
 		case <-time.After(txn.opTimeout):
-			if _, err := txn.checkIfOpen(); err != nil {
+			if err := txn.verifyOpen(); err != nil {
 				return err
 			}
 			return newError(TimeoutError, "Failed to get the semaphore to register the send/ack operation")
@@ -154,23 +166,22 @@ func (txn *transaction) registerSendOrAckOp() error {
 
 func (txn *transaction) endSendOrAckOp(err error) {
 	if err != nil {
-		atomic.StoreInt32((*int32)(&txn.state), int32(TxnError))
+		txn.state.Store(int32(TxnError))
 	}
-	if atomic.AddInt32(&txn.opsCount, -1) == 0 {
-		//This means there are not pending send/ack requests
+	if txn.opsCount.Add(-1) == 0 {
+		// This means there are no pending send/ack requests
 		txn.opsFlow <- true
 	}
 }
 
 func (txn *transaction) registerProducerTopic(topic string) error {
-	isOpen, err := txn.checkIfOpen()
-	if !isOpen {
+	if err := txn.verifyOpen(); err != nil {
 		return err
 	}
 	_, ok := txn.registerPartitions[topic]
 	if !ok {
-		txn.Lock()
-		defer txn.Unlock()
+		txn.mu.Lock()
+		defer txn.mu.Unlock()
 		if _, ok = txn.registerPartitions[topic]; !ok {
 			err := txn.tcClient.addPublishPartitionToTxn(&txn.txnID, []string{topic})
 			if err != nil {
@@ -183,8 +194,7 @@ func (txn *transaction) registerProducerTopic(topic string) error {
 }
 
 func (txn *transaction) registerAckTopic(topic string, subName string) error {
-	isOpen, err := txn.checkIfOpen()
-	if !isOpen {
+	if err := txn.verifyOpen(); err != nil {
 		return err
 	}
 	sub := subscription{
@@ -193,8 +203,8 @@ func (txn *transaction) registerAckTopic(topic string, subName string) error {
 	}
 	_, ok := txn.registerAckSubscriptions[sub]
 	if !ok {
-		txn.Lock()
-		defer txn.Unlock()
+		txn.mu.Lock()
+		defer txn.mu.Unlock()
 		if _, ok = txn.registerAckSubscriptions[sub]; !ok {
 			err := txn.tcClient.addSubscriptionToTxn(&txn.txnID, topic, subName)
 			if err != nil {
@@ -210,14 +220,15 @@ func (txn *transaction) GetTxnID() TxnID {
 	return txn.txnID
 }
 
-func (txn *transaction) checkIfOpen() (bool, error) {
-	if txn.state == TxnOpen {
-		return true, nil
+func (txn *transaction) verifyOpen() error {
+	txnState := txn.state.Load()
+	if txnState != int32(TxnOpen) {
+		return newError(InvalidStatus, txnStateErrorMessage(TxnOpen, TxnState(txnState)))
 	}
-	return false, newError(InvalidStatus, "Expect transaction state is TxnOpen but "+txn.state.string())
+	return nil
 }
 
-func (state TxnState) string() string {
+func (state TxnState) String() string {
 	switch state {
 	case TxnOpen:
 		return "TxnOpen"
@@ -237,3 +248,8 @@ func (state TxnState) string() string {
 		return "Unknown"
 	}
 }
+
+//nolint:unparam
+func txnStateErrorMessage(expected, actual TxnState) string {
+	return fmt.Sprintf("Expected transaction state: %s, actual: %s", expected, actual)
+}


Test patch:
diff --git a/pulsar/transaction_test.go b/pulsar/transaction_test.go
index eb88c70..75b36ea 100644
--- a/pulsar/transaction_test.go
+++ b/pulsar/transaction_test.go
@@ -29,7 +29,9 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestTCClient(t *testing.T) {
+const txnTimeout = 10 * time.Minute
+
+func TestTxn_TCClient(t *testing.T) {
 	//1. Prepare: create PulsarClient and init transaction coordinator client.
 	topic := newTopicName()
 	sub := "my-sub"
@@ -52,13 +54,13 @@ func TestTCClient(t *testing.T) {
 	stats, err := transactionStats(id1)
 	assert.NoError(t, err)
 	assert.Equal(t, "OPEN", stats["status"])
-	producedPartitions := stats["producedPartitions"].(map[string]interface{})
-	ackedPartitions := stats["ackedPartitions"].(map[string]interface{})
+	producedPartitions := stats["producedPartitions"].(map[string]any)
+	ackedPartitions := stats["ackedPartitions"].(map[string]any)
 	_, ok := producedPartitions[topic]
 	assert.True(t, ok)
 	temp, ok := ackedPartitions[topic]
 	assert.True(t, ok)
-	subscriptions := temp.(map[string]interface{})
+	subscriptions := temp.(map[string]any)
 	_, ok = subscriptions[sub]
 	assert.True(t, ok)
 	//5. Test End transaction
@@ -78,12 +80,12 @@ func TestTCClient(t *testing.T) {
 	} else {
 		assert.Equal(t, err.Error(), "http error status code: 404")
 	}
-	defer consumer.Close()
-	defer tc.close()
-	defer client.Close()
+	consumer.Close()
+	tc.close()
+	client.Close()
 }
 
-//Test points:
+// Test points:
 //	1. Abort and commit txn.
 //		1. Do nothing, just open a transaction and then commit it or abort it.
 //		The operations of committing/aborting txn should success at the first time and fail at the second time.
@@ -96,8 +98,8 @@ func TestTCClient(t *testing.T) {
 //		1. Register ack topic and send topic, and call http request to get the stats of the transaction
 //		to do verification.
 
-// TestTxnImplCommitOrAbort Test abort and commit txn
-func TestTxnImplCommitOrAbort(t *testing.T) {
+// Test abort and commit txn
+func TestTxn_ImplCommitOrAbort(t *testing.T) {
 	tc, _ := createTcClient(t)
 	//1. Open a transaction and then commit it.
 	//The operations of committing txn1 should success at the first time and fail at the second time.
@@ -105,20 +107,20 @@ func TestTxnImplCommitOrAbort(t *testing.T) {
 	err := txn1.Commit(context.Background())
 	require.Nil(t, err, fmt.Sprintf("Failed to commit the transaction %d:%d\n", txn1.txnID.MostSigBits,
 		txn1.txnID.LeastSigBits))
-	txn1.state = TxnOpen
+	txn1.state.Store(int32(TxnOpen))
 	txn1.opsFlow <- true
 	err = txn1.Commit(context.Background())
 	assert.Equal(t, err.(*Error).Result(), TransactionNoFoundError)
 	assert.Equal(t, txn1.GetState(), TxnError)
 	//2. Open a transaction and then abort it.
 	//The operations of aborting txn2 should success at the first time and fail at the second time.
-	id2, err := tc.newTransaction(time.Hour)
-	require.Nil(t, err, "Failed to new a transaction")
-	txn2 := newTransaction(*id2, tc, time.Hour)
+	id2, err := tc.newTransaction(txnTimeout)
+	require.Nil(t, err, "Failed to create a transaction")
+	txn2 := newTransaction(*id2, tc, txnTimeout)
 	err = txn2.Abort(context.Background())
 	require.Nil(t, err, fmt.Sprintf("Failed to abort the transaction %d:%d\n",
 		id2.MostSigBits, id2.LeastSigBits))
-	txn2.state = TxnOpen
+	txn2.state.Store(int32(TxnOpen))
 	txn2.opsFlow <- true
 	err = txn2.Abort(context.Background())
 	assert.Equal(t, err.(*Error).Result(), TransactionNoFoundError)
@@ -129,8 +131,8 @@ func TestTxnImplCommitOrAbort(t *testing.T) {
 	assert.Equal(t, err.(*Error).Result(), InvalidStatus)
 }
 
-// TestRegisterOpAndEndOp Test the internal API including the registerSendOrAckOp and endSendOrAckOp.
-func TestRegisterOpAndEndOp(t *testing.T) {
+// Test the internal API including the registerSendOrAckOp and endSendOrAckOp.
+func TestTxn_RegisterOpAndEndOp(t *testing.T) {
 	tc, _ := createTcClient(t)
 	//1. Register 4 operation but only end 3 operations, the transaction can not be committed or aborted.
 	res := registerOpAndEndOp(t, tc, 4, 3, nil, true)
@@ -151,8 +153,8 @@ func TestRegisterOpAndEndOp(t *testing.T) {
 	assert.Equal(t, res.(*Error).Result(), InvalidStatus)
 }
 
-// TestRegisterTopic Test the internal API, registerAckTopic and registerProducerTopic
-func TestRegisterTopic(t *testing.T) {
+// Test the internal API, registerAckTopic and registerProducerTopic
+func TestTxn_RegisterTopic(t *testing.T) {
 	//1. Prepare: create PulsarClient and init transaction coordinator client.
 	topic := newTopicName()
 	sub := "my-sub"
@@ -172,11 +174,11 @@ func TestRegisterTopic(t *testing.T) {
 	//4. Call http request to get the stats of the transaction to do verification.
 	stats2, err := transactionStats(&txn.txnID)
 	assert.NoError(t, err)
-	topics := stats2["producedPartitions"].(map[string]interface{})
-	subTopics := stats2["ackedPartitions"].(map[string]interface{})
+	topics := stats2["producedPartitions"].(map[string]any)
+	subTopics := stats2["ackedPartitions"].(map[string]any)
 	assert.NotNil(t, topics[topic])
 	assert.NotNil(t, subTopics[topic])
-	subs := subTopics[topic].(map[string]interface{})
+	subs := subTopics[topic].(map[string]any)
 	assert.NotNil(t, subs[sub])
 }
 
@@ -198,9 +200,9 @@ func registerOpAndEndOp(t *testing.T, tc *transactionCoordinatorClient, rp int,
 }
 
 func createTxn(tc *transactionCoordinatorClient, t *testing.T) *transaction {
-	id, err := tc.newTransaction(time.Hour)
-	require.Nil(t, err, "Failed to new a transaction.")
-	return newTransaction(*id, tc, time.Hour)
+	id, err := tc.newTransaction(txnTimeout)
+	require.Nil(t, err, "Failed to create a new transaction.")
+	return newTransaction(*id, tc, txnTimeout)
 }
 
 // createTcClient Create a transaction coordinator client to send request
@@ -216,7 +218,7 @@ func createTcClient(t *testing.T) (*transactionCoordinatorClient, *client) {
 	return c.(*client).tcClient, c.(*client)
 }
 
-// TestConsumeAndProduceWithTxn is a test function that validates the behavior of producing and consuming
+// Validate the behavior of producing and consuming
 // messages with and without transactions. It consists of the following steps:
 //
 // 1. Prepare: Create a PulsarClient and initialize the transaction coordinator client.
@@ -230,7 +232,7 @@ func createTcClient(t *testing.T) (*transactionCoordinatorClient, *client) {
 //
 // The test ensures that the consumer can only receive messages sent with a transaction after it is committed,
 // and that it can always receive messages sent without a transaction.
-func TestConsumeAndProduceWithTxn(t *testing.T) {
+func TestTxn_ConsumeAndProduce(t *testing.T) {
 	// Step 1: Prepare - Create PulsarClient and initialize the transaction coordinator client.
 	topic := newTopicName()
 	sub := "my-sub"
@@ -249,7 +251,7 @@ func TestConsumeAndProduceWithTxn(t *testing.T) {
 	// Step 3: Open a transaction, send 10 messages with the transaction and 10 messages without the transaction.
 	// Expectation: We can receive the 10 messages sent without a transaction and
 	// cannot receive the 10 messages sent with the transaction.
-	txn, err := client.NewTransaction(time.Hour)
+	txn, err := client.NewTransaction(txnTimeout)
 	require.Nil(t, err)
 	for i := 0; i < 10; i++ {
 		_, err = producer.Send(context.Background(), &ProducerMessage{
@@ -289,7 +291,7 @@ func TestConsumeAndProduceWithTxn(t *testing.T) {
 	// Acknowledge the rest of the 10 messages with the transaction.
 	// Expectation: After committing the transaction, all messages of the subscription will be acknowledged.
 	_ = txn.Commit(context.Background())
-	txn, err = client.NewTransaction(time.Hour)
+	txn, err = client.NewTransaction(txnTimeout)
 	require.Nil(t, err)
 	for i := 0; i < 9; i++ {
 		msg, _ := consumer.Receive(context.Background())
@@ -307,7 +309,7 @@ func TestConsumeAndProduceWithTxn(t *testing.T) {
 	// Create a goroutine to attempt receiving a message and send it to the 'done1' channel.
 	done2 := make(chan Message)
 	go func() {
-		consumer.Receive(context.Background())
+		_, _ = consumer.Receive(context.Background())
 		close(done2)
 	}()
 
@@ -323,7 +325,7 @@ func TestConsumeAndProduceWithTxn(t *testing.T) {
 	producer.Close()
 }
 
-func TestAckAndSendWithTxn(t *testing.T) {
+func TestTxn_AckAndSend(t *testing.T) {
 	// Prepare: Create PulsarClient and initialize the transaction coordinator client.
 	sourceTopic := newTopicName()
 	sinkTopic := newTopicName()
@@ -357,7 +359,7 @@ func TestAckAndSendWithTxn(t *testing.T) {
 	}
 
 	// Open a transaction and consume messages from the source topic while sending messages to the sink topic.
-	txn, err := client.NewTransaction(time.Hour)
+	txn, err := client.NewTransaction(txnTimeout)
 	require.Nil(t, err)
 
 	for i := 0; i < 10; i++ {
@@ -393,7 +395,7 @@ func TestAckAndSendWithTxn(t *testing.T) {
 	sinkProducer.Close()
 }
 
-func TestTransactionAbort(t *testing.T) {
+func TestTxn_TransactionAbort(t *testing.T) {
 	// Prepare: Create PulsarClient and initialize the transaction coordinator client.
 	topic := newTopicName()
 	sub := "my-sub"
@@ -410,7 +412,7 @@ func TestTransactionAbort(t *testing.T) {
 	})
 
 	// Open a transaction and send 10 messages with the transaction.
-	txn, err := client.NewTransaction(time.Hour)
+	txn, err := client.NewTransaction(txnTimeout)
 	require.Nil(t, err)
 
 	for i := 0; i < 10; i++ {
@@ -449,7 +451,7 @@ func consumerShouldNotReceiveMessage(t *testing.T, consumer Consumer) {
 	}
 }
 
-func TestTransactionAckChunkMessage(t *testing.T) {
+func TestTxn_AckChunkMessage(t *testing.T) {
 	topic := newTopicName()
 	sub := "my-sub"
 
@@ -457,7 +459,7 @@ func TestTransactionAckChunkMessage(t *testing.T) {
 	_, client := createTcClient(t)
 
 	// Create transaction and register the send operation.
-	txn, err := client.NewTransaction(time.Hour)
+	txn, err := client.NewTransaction(txnTimeout)
 	require.Nil(t, err)
 
 	// Create a producer with chunking enabled to send a large message that will be split into chunks.
@@ -487,13 +489,13 @@ func TestTransactionAckChunkMessage(t *testing.T) {
 	})
 	require.NoError(t, err)
 	_, ok := msgID.(*chunkMessageID)
-	require.True(t, ok)
+	require.True(t, ok, fmt.Sprintf("Expected message ID of type *chunkMessageID, got type %T", msgID))
 
 	err = txn.Commit(context.Background())
 	require.Nil(t, err)
 
 	// Receive the message using a new transaction and ack it.
-	txn2, err := client.NewTransaction(time.Hour)
+	txn2, err := client.NewTransaction(txnTimeout)
 	require.Nil(t, err)
 	message, err := consumer.Receive(context.Background())
 	require.Nil(t, err)
@@ -501,7 +503,7 @@ func TestTransactionAckChunkMessage(t *testing.T) {
 	err = consumer.AckWithTxn(message, txn2)
 	require.Nil(t, err)
 
-	txn2.Abort(context.Background())
+	_ = txn2.Abort(context.Background())
 
 	// Close the consumer to simulate reconnection and receive the same message again.
 	consumer.Close()
@@ -518,7 +520,7 @@ func TestTransactionAckChunkMessage(t *testing.T) {
 	require.NotNil(t, message)
 
 	// Create a new transaction and ack the message again.
-	txn3, err := client.NewTransaction(time.Hour)
+	txn3, err := client.NewTransaction(txnTimeout)
 	require.Nil(t, err)
 
 	err = consumer.AckWithTxn(message, txn3)
@@ -541,7 +543,7 @@ func TestTransactionAckChunkMessage(t *testing.T) {
 	consumerShouldNotReceiveMessage(t, consumer)
 }
 
-func TestTxnConnReconnect(t *testing.T) {
+func TestTxn_ConnReconnect(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
@@ -593,3 +595,8 @@ func TestTxnConnReconnect(t *testing.T) {
 	err = txn.Commit(context.Background())
 	assert.NoError(t, err)
 }
+
+func TestTxn_txnStateErrorMessage(t *testing.T) {
+	expected := "Expected transaction state: TxnOpen, actual: TxnTimeout"
+	assert.Equal(t, expected, txnStateErrorMessage(TxnOpen, TxnTimeout))
+}