Repository: apache/pulsar-client-go
Pull request #1002 (2023-04-05 06:28:18)

[feat][txn]Implement transactional consumer/producer API
Master Issue:https://github.com/apache/pulsar-client-go/issues/932
### Motivation
Implement transaction coordinator client.
### Modifications
1. Implement transaction coordinator
2. implement transactionImpl
3. Implement transaction in producer and consumer API

### Verifying this change

- [ ] Make sure that the change passes the CI checks.

*(Please pick either of the following options)*

This change is a trivial rework / code cleanup without any test coverage.

*(or)*

This change is already covered by existing tests, such as *(please describe tests)*.

*(or)*

This change added tests and can be verified as follows:

*(example:)*
  - *Added integration tests for end-to-end deployment with large payloads (10MB)*
  - *Extended integration test for recovery after broker failure*

### Does this pull request potentially affect one of the following parts:

*If `yes` was chosen, please highlight the changes*

  - Dependencies (does it add or upgrade a dependency): (yes / no)
  - The public API: (yes / no)
  - The schema: (yes / no / don't know)
  - The default values of configurations: (yes / no)
  - The wire protocol: (yes / no)

### Documentation

  - Does this pull request introduce a new feature? (yes / no)
  - If yes, how is the feature documented? (not applicable / docs / GoDocs / not documented)
  - If a feature is not applicable for documentation, explain why?
  - If a feature is not documented yet in this PR, please create a followup issue for adding the documentation


shibd: Is this the last PR about transactions? It's better to add some sample code to examples.
liangyepianzhou: >Is this the last PR about transactions? It's better to add some sample code to examples.

Yes, this is the last PR concerning transactions. In the tests I've added, I've only used the API to perform the tests, which can serve as an initial example. Adding more detailed examples to the documentation might be more appropriate?
shibd: > Adding more detailed examples to the documentation might be more appropriate?

Yes, the documentation can be described step by step, and complete examples are placed in the examples folder. We can push in a later PR.

Patch:
diff --git a/.github/workflows/bot.yml b/.github/workflows/bot.yml
index eb86c56..4641eb9 100644
--- a/.github/workflows/bot.yml
+++ b/.github/workflows/bot.yml
@@ -1,25 +1,34 @@
-name: Bot tests
+# 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.
+
+name: Pulsar Bot
 on:
   issue_comment:
     types: [created]
-  pull_request:
-    types: [closed]
 
 jobs:
-  bot:
-    runs-on: ubuntu-latest
-    timeout-minutes: 120
-
+  pulsarbot:
+    runs-on: ubuntu-20.04
+    timeout-minutes: 10
+    if: github.event_name == 'issue_comment' && contains(github.event.comment.body, '/pulsarbot')
     steps:
-      - name: checkout
-        uses: actions/checkout@v2
-        with:
-          fetch-depth: 100
-          ref: ${{ github.event.pull_request.head.sha }}
       - name: Execute pulsarbot command
-        id:   pulsarbot
-        if: github.event_name == 'issue_comment' && startsWith(github.event.comment.body, '/pulsarbot')
+        id: pulsarbot
         env:
-          GITHUB_TOKEN: ${{ secrets.GO_CLIENT_BOT_TOKEN }}
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           BOT_TARGET_REPOSITORY: apache/pulsar-client-go
         uses: apache/pulsar-test-infra/pulsarbot@master
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
new file mode 100644
index 0000000..7c946b7
--- /dev/null
+++ b/.github/workflows/ci.yml
@@ -0,0 +1,47 @@
+# 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.
+
+name: CI
+on: [pull_request]
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - uses: actions/setup-go@v3
+      - run: make build
+
+  lint:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - run: make lint
+
+  integration-tests:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        go-version: [1.16, 1.17, 1.18, 1.19]
+    steps:
+      - uses: actions/checkout@v3
+      - name: clean docker cache
+        run: docker rmi $(docker images -q) -f && df -h
+      - uses: actions/setup-go@v3
+        with:
+          go-version: ${{ matrix.go-version }}
+      - name: Run Tests
+        run: make test GO_VERSION=${{ matrix.go-version }}
diff --git a/.github/workflows/project.yml b/.github/workflows/project.yml
deleted file mode 100644
index a52a4be..0000000
--- a/.github/workflows/project.yml
+++ /dev/null
@@ -1,39 +0,0 @@
-name: CI
-on: [pull_request]
-jobs:
-  build:
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v3
-      - uses: actions/setup-go@v3
-      - name: build
-        run: make build
-
-  lint:
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v3
-      - uses: golangci/golangci-lint-action@v3
-        with:
-          version: v1.50.1
-
-  integration-tests:
-    runs-on: ubuntu-latest
-    strategy:
-      matrix:
-        go-version: [1.16, 1.17, 1.18, 1.19]
-    steps:
-      - name: clean docker cache
-        run: |
-          docker rmi $(docker images -q) -f
-          df -h
-
-      - uses: actions/setup-go@v3
-        with:
-          go-version: ${{ matrix.go-version }}
-
-      - name: Check out code into the Go module directory
-        uses: actions/checkout@v3
-
-      - name: Run Tests
-        run: make test GO_VERSION=${{ matrix.go-version }}
diff --git a/.golangci.yml b/.golangci.yml
index 2dc120a..1f1b42c 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -1,6 +1,22 @@
-# Make sure golangci-lint is installed.
-# https://github.com/golangci/golangci-lint#install
-# We can execute `golangci-lint run` for code checking.
+# 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.
+
+# Run `make lint` from the root path of this project to check code with golangci-lint.
+
 run:
   deadline: 6m
 
diff --git a/Makefile b/Makefile
index 231c05e..433db8c 100644
--- a/Makefile
+++ b/Makefile
@@ -23,12 +23,19 @@ PULSAR_IMAGE = apachepulsar/pulsar:$(PULSAR_VERSION)
 GO_VERSION ?= 1.18
 GOLANG_IMAGE = golang:$(GO_VERSION)
 
+# Golang standard bin directory.
+GOPATH ?= $(shell go env GOPATH)
+GOROOT ?= $(shell go env GOROOT)
+
 build:
 	go build ./pulsar
 	go build -o bin/pulsar-perf ./perf
 
-lint:
-	golangci-lint run
+lint: bin/golangci-lint
+	bin/golangci-lint run
+
+bin/golangci-lint:
+	GOBIN=$(shell pwd)/bin go install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.51.2
 
 container:
 	docker build -t ${IMAGE_NAME} --build-arg GOLANG_IMAGE="${GOLANG_IMAGE}" \
diff --git a/go.mod b/go.mod
index 3e7299b..69862b2 100644
--- a/go.mod
+++ b/go.mod
@@ -27,7 +27,7 @@ require (
 	golang.org/x/mod v0.5.1
 	golang.org/x/oauth2 v0.0.0-20210402161424-2e8d93401602
 	golang.org/x/time v0.0.0-20191024005414-555d28b269f0
-	google.golang.org/protobuf v1.26.0
+	google.golang.org/protobuf v1.30.0 // indirect
 )
 
 require (
diff --git a/go.sum b/go.sum
index 528c458..3a39874 100644
--- a/go.sum
+++ b/go.sum
@@ -553,6 +553,8 @@ google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlba
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
 google.golang.org/protobuf v1.26.0 h1:bxAC2xTBsZGibn2RTntX0oH50xLsqy1OxA9tTL3p/lk=
 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
+google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cng=
+google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
diff --git a/pulsar/client.go b/pulsar/client.go
index bc3f4f5..7e6725d 100644
--- a/pulsar/client.go
+++ b/pulsar/client.go
@@ -184,6 +184,17 @@ type Client interface {
 	// {@link Consumer} or {@link Producer} instances directly on a particular partition.
 	TopicPartitions(topic string) ([]string, error)
 
+	// NewTransaction creates a new Transaction instance.
+	//
+	// This function is used to initiate a new transaction for performing
+	// atomic operations on the message broker. It returns a Transaction
+	// object that can be used to produce, consume and commit messages in a
+	// transactional manner.
+	//
+	// In case of any errors while creating the transaction, an error will
+	// be returned.
+	NewTransaction(duration time.Duration) (Transaction, error)
+
 	// Close Closes the Client and free associated resources
 	Close()
 }
diff --git a/pulsar/client_impl.go b/pulsar/client_impl.go
index 5322597..9b44987 100644
--- a/pulsar/client_impl.go
+++ b/pulsar/client_impl.go
@@ -195,6 +195,14 @@ func newClient(options ClientOptions) (Client, error) {
 	return c, nil
 }
 
+func (c *client) NewTransaction(timeout time.Duration) (Transaction, error) {
+	id, err := c.tcClient.newTransaction(timeout)
+	if err != nil {
+		return nil, err
+	}
+	return newTransaction(*id, c.tcClient, timeout), nil
+}
+
 func (c *client) CreateProducer(options ProducerOptions) (Producer, error) {
 	producer, err := newProducer(c, &options)
 	if err == nil {
diff --git a/pulsar/consumer.go b/pulsar/consumer.go
index 64a096d..3ef72c7 100644
--- a/pulsar/consumer.go
+++ b/pulsar/consumer.go
@@ -265,6 +265,9 @@ type Consumer interface {
 	// AckID the consumption of a single message, identified by its MessageID
 	AckID(MessageID) error
 
+	// AckWithTxn the consumption of a single message with a transaction
+	AckWithTxn(Message, Transaction) error
+
 	// AckCumulative the reception of all the messages in the stream up to (and including)
 	// the provided message.
 	AckCumulative(msg Message) error
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
index fd7fa57..07f38c3 100644
--- a/pulsar/consumer_impl.go
+++ b/pulsar/consumer_impl.go
@@ -38,6 +38,7 @@ type acker interface {
 	// AckID does not handle errors returned by the Broker side, so no need to wait for doneCh to finish.
 	AckID(id MessageID) error
 	AckIDWithResponse(id MessageID) error
+	AckIDWithTxn(msgID MessageID, txn Transaction) error
 	AckIDCumulative(msgID MessageID) error
 	AckIDWithResponseCumulative(msgID MessageID) error
 	NackID(id MessageID)
@@ -478,6 +479,15 @@ func (c *consumer) Receive(ctx context.Context) (message Message, err error) {
 	}
 }
 
+func (c *consumer) AckWithTxn(msg Message, txn Transaction) error {
+	msgID := msg.ID()
+	if err := c.checkMsgIDPartition(msgID); err != nil {
+		return err
+	}
+
+	return c.consumers[msgID.PartitionIdx()].AckIDWithTxn(msgID, txn)
+}
+
 // Chan return the message chan to users
 func (c *consumer) Chan() <-chan ConsumerMessage {
 	return c.messageCh
diff --git a/pulsar/consumer_multitopic.go b/pulsar/consumer_multitopic.go
index 8108c29..f6630dd 100644
--- a/pulsar/consumer_multitopic.go
+++ b/pulsar/consumer_multitopic.go
@@ -143,6 +143,23 @@ func (c *multiTopicConsumer) AckID(msgID MessageID) error {
 	return mid.consumer.AckID(msgID)
 }
 
+// AckWithTxn the consumption of a single message with a transaction
+func (c *multiTopicConsumer) AckWithTxn(msg Message, txn Transaction) error {
+	msgID := msg.ID()
+	if !checkMessageIDType(msgID) {
+		c.log.Warnf("invalid message id type %T", msgID)
+		return errors.New("invalid message id type in multi_consumer")
+	}
+	mid := toTrackingMessageID(msgID)
+
+	if mid.consumer == nil {
+		c.log.Warnf("unable to ack messageID=%+v can not determine topic", msgID)
+		return errors.New("unable to ack message because consumer is nil")
+	}
+
+	return mid.consumer.AckIDWithTxn(msgID, txn)
+}
+
 // AckCumulative the reception of all the messages in the stream up to (and including)
 // the provided message
 func (c *multiTopicConsumer) AckCumulative(msg Message) error {
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index fb77d0d..6e241d8 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -417,6 +417,118 @@ func (pc *partitionConsumer) Unsubscribe() error {
 	return req.err
 }
 
+// ackIDCommon handles common logic for acknowledging messages with or without transactions.
+// withTxn should be set to true when dealing with transactions.
+func (pc *partitionConsumer) ackIDCommon(msgID MessageID, withResponse bool, txn Transaction) error {
+	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
+		pc.log.WithField("state", state).Error("Failed to ack by closing or closed consumer")
+		return errors.New("consumer state is closed")
+	}
+
+	if cmid, ok := msgID.(*chunkMessageID); ok {
+		return pc.unAckChunksTracker.ack(cmid)
+	}
+
+	trackingID := toTrackingMessageID(msgID)
+
+	if trackingID != nil && trackingID.ack() {
+		// All messages in the same batch have been acknowledged, we only need to acknowledge the
+		// MessageID that represents the entry that stores the whole batch
+		trackingID = &trackingMessageID{
+			messageID: &messageID{
+				ledgerID: trackingID.ledgerID,
+				entryID:  trackingID.entryID,
+			},
+		}
+		pc.metrics.AcksCounter.Inc()
+		pc.metrics.ProcessingTime.Observe(float64(time.Now().UnixNano()-trackingID.receivedTime.UnixNano()) / 1.0e9)
+	} else if !pc.options.enableBatchIndexAck {
+		return nil
+	}
+
+	var err error
+	if withResponse {
+		if txn != nil {
+			ackReq := pc.sendIndividualAckWithTxn(trackingID, txn.(*transaction))
+			<-ackReq.doneCh
+			err = ackReq.err
+		} else {
+			ackReq := pc.sendIndividualAck(trackingID)
+			<-ackReq.doneCh
+			err = ackReq.err
+		}
+	} else {
+		pc.ackGroupingTracker.add(trackingID)
+	}
+	pc.options.interceptors.OnAcknowledge(pc.parentConsumer, msgID)
+	return err
+}
+
+// AckIDWithTxn acknowledges the consumption of a message with transaction.
+func (pc *partitionConsumer) AckIDWithTxn(msgID MessageID, txn Transaction) error {
+	return pc.ackIDCommon(msgID, true, txn)
+}
+
+// ackID acknowledges the consumption of a message and optionally waits for response from the broker.
+func (pc *partitionConsumer) ackID(msgID MessageID, withResponse bool) error {
+	return pc.ackIDCommon(msgID, withResponse, nil)
+}
+
+func (pc *partitionConsumer) internalAckWithTxn(req *ackWithTxnRequest) {
+	defer close(req.doneCh)
+	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
+		pc.log.WithField("state", state).Error("Failed to ack by closing or closed consumer")
+		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.")
+		req.err = newError(InvalidStatus, "Failed to ack by a non-open transaction.")
+		return
+	}
+	msgID := req.msgID
+
+	messageIDs := make([]*pb.MessageIdData, 1)
+	messageIDs[0] = &pb.MessageIdData{
+		LedgerId: proto.Uint64(uint64(msgID.ledgerID)),
+		EntryId:  proto.Uint64(uint64(msgID.entryID)),
+	}
+	if pc.options.enableBatchIndexAck && msgID.tracker != nil {
+		ackSet := msgID.tracker.toAckSet()
+		if ackSet != nil {
+			messageIDs[0].AckSet = ackSet
+		}
+	}
+
+	reqID := pc.client.rpcClient.NewRequestID()
+	txnID := req.Transaction.GetTxnID()
+	cmdAck := &pb.CommandAck{
+		ConsumerId:     proto.Uint64(pc.consumerID),
+		MessageId:      messageIDs,
+		AckType:        pb.CommandAck_Individual.Enum(),
+		TxnidMostBits:  proto.Uint64(txnID.MostSigBits),
+		TxnidLeastBits: proto.Uint64(txnID.LeastSigBits),
+	}
+
+	if err := req.Transaction.registerAckTopic(pc.options.topic, pc.options.subscription); err != nil {
+		req.err = err
+		return
+	}
+
+	if err := req.Transaction.registerSendOrAckOp(); err != nil {
+		req.err = err
+		return
+	}
+
+	cmdAck.RequestId = proto.Uint64(reqID)
+	_, err := pc.client.rpcClient.RequestOnCnx(pc._getConn(), reqID, pb.BaseCommand_ACK, cmdAck)
+	if err != nil {
+		pc.log.WithError(err).Error("Ack with response error")
+	}
+	req.Transaction.endSendOrAckOp(err)
+	req.err = err
+}
+
 func (pc *partitionConsumer) internalUnsubscribe(unsub *unsubscribeRequest) {
 	defer close(unsub.doneCh)
 
@@ -488,47 +600,6 @@ func (pc *partitionConsumer) requestGetLastMessageID() (*trackingMessageID, erro
 	return convertToMessageID(id), nil
 }
 
-func (pc *partitionConsumer) ackID(msgID MessageID, withResponse bool) error {
-	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
-		pc.log.WithField("state", state).Error("Failed to ack by closing or closed consumer")
-		return errors.New("consumer state is closed")
-	}
-
-	if cmid, ok := msgID.(*chunkMessageID); ok {
-		return pc.unAckChunksTracker.ack(cmid)
-	}
-
-	trackingID := toTrackingMessageID(msgID)
-
-	if trackingID != nil && trackingID.ack() {
-		// All messages in the same batch have been acknowledged, we only need to acknowledge the
-		// MessageID that represents the entry that stores the whole batch
-		trackingID = &trackingMessageID{
-			messageID: &messageID{
-				ledgerID: trackingID.ledgerID,
-				entryID:  trackingID.entryID,
-			},
-		}
-		pc.metrics.AcksCounter.Inc()
-		pc.metrics.ProcessingTime.Observe(float64(time.Now().UnixNano()-trackingID.receivedTime.UnixNano()) / 1.0e9)
-	} else if !pc.options.enableBatchIndexAck {
-		return nil
-	}
-
-	var ackReq *ackRequest
-	if withResponse {
-		ackReq := pc.sendIndividualAck(trackingID)
-		<-ackReq.doneCh
-	} else {
-		pc.ackGroupingTracker.add(trackingID)
-	}
-	pc.options.interceptors.OnAcknowledge(pc.parentConsumer, msgID)
-	if ackReq == nil {
-		return nil
-	}
-	return ackReq.err
-}
-
 func (pc *partitionConsumer) sendIndividualAck(msgID MessageID) *ackRequest {
 	ackReq := &ackRequest{
 		doneCh:  make(chan struct{}),
@@ -539,6 +610,17 @@ func (pc *partitionConsumer) sendIndividualAck(msgID MessageID) *ackRequest {
 	return ackReq
 }
 
+func (pc *partitionConsumer) sendIndividualAckWithTxn(msgID MessageID, txn *transaction) *ackWithTxnRequest {
+	ackReq := &ackWithTxnRequest{
+		Transaction: txn,
+		doneCh:      make(chan struct{}),
+		ackType:     individualAck,
+		msgID:       *msgID.(*trackingMessageID),
+	}
+	pc.eventsCh <- ackReq
+	return ackReq
+}
+
 func (pc *partitionConsumer) AckIDWithResponse(msgID MessageID) error {
 	if !checkMessageIDType(msgID) {
 		pc.log.Errorf("invalid message id type %T", msgID)
@@ -1389,6 +1471,14 @@ type ackRequest struct {
 	err     error
 }
 
+type ackWithTxnRequest struct {
+	doneCh      chan struct{}
+	msgID       trackingMessageID
+	Transaction *transaction
+	ackType     int
+	err         error
+}
+
 type unsubscribeRequest struct {
 	doneCh chan struct{}
 	err    error
@@ -1444,6 +1534,8 @@ func (pc *partitionConsumer) runEventsLoop() {
 			switch v := i.(type) {
 			case *ackRequest:
 				pc.internalAck(v)
+			case *ackWithTxnRequest:
+				pc.internalAckWithTxn(v)
 			case []*pb.MessageIdData:
 				pc.internalAckList(v)
 			case *redeliveryRequest:
@@ -1525,6 +1617,11 @@ func (pc *partitionConsumer) reconnectToBroker() {
 		maxRetry = int(*pc.options.maxReconnectToBroker)
 	}
 
+	var (
+		delayReconnectTime time.Duration
+		defaultBackoff     = internal.DefaultBackoff{}
+	)
+
 	for maxRetry != 0 {
 		if pc.getConsumerState() != consumerReady {
 			// Consumer is already closing
@@ -1532,11 +1629,6 @@ func (pc *partitionConsumer) reconnectToBroker() {
 			return
 		}
 
-		var (
-			delayReconnectTime time.Duration
-			defaultBackoff     = internal.DefaultBackoff{}
-		)
-
 		if pc.options.backoffPolicy == nil {
 			delayReconnectTime = defaultBackoff.Next()
 		} else {
diff --git a/pulsar/consumer_regex.go b/pulsar/consumer_regex.go
index 2520af5..79e2293 100644
--- a/pulsar/consumer_regex.go
+++ b/pulsar/consumer_regex.go
@@ -193,6 +193,24 @@ func (c *regexConsumer) AckID(msgID MessageID) error {
 	return mid.consumer.AckID(msgID)
 }
 
+// AckID the consumption of a single message, identified by its MessageID
+func (c *regexConsumer) AckWithTxn(msg Message, txn Transaction) error {
+	msgID := msg.ID()
+	if !checkMessageIDType(msgID) {
+		c.log.Warnf("invalid message id type %T", msgID)
+		return fmt.Errorf("invalid message id type %T", msgID)
+	}
+
+	mid := toTrackingMessageID(msgID)
+
+	if mid.consumer == nil {
+		c.log.Warnf("unable to ack messageID=%+v can not determine topic", msgID)
+		return errors.New("consumer is nil in consumer_regex")
+	}
+
+	return mid.consumer.AckIDWithTxn(msgID, txn)
+}
+
 // AckCumulative the reception of all the messages in the stream up to (and including)
 // the provided message.
 func (c *regexConsumer) AckCumulative(msg Message) error {
diff --git a/pulsar/internal/batch_builder.go b/pulsar/internal/batch_builder.go
index 649aba4..6df3a61 100644
--- a/pulsar/internal/batch_builder.go
+++ b/pulsar/internal/batch_builder.go
@@ -51,6 +51,9 @@ type BatchBuilder interface {
 		payload []byte,
 		callback interface{}, replicateTo []string, deliverAt time.Time,
 		schemaVersion []byte, multiSchemaEnabled bool,
+		useTxn bool,
+		mostSigBits uint64,
+		leastSigBits uint64,
 	) bool
 
 	// Flush all the messages buffered in the client and wait until all messages have been successfully persisted.
@@ -185,6 +188,7 @@ func (bc *batchContainer) Add(
 	payload []byte,
 	callback interface{}, replicateTo []string, deliverAt time.Time,
 	schemaVersion []byte, multiSchemaEnabled bool,
+	useTxn bool, mostSigBits uint64, leastSigBits uint64,
 ) bool {
 
 	if replicateTo != nil && bc.numMessages != 0 {
@@ -223,6 +227,10 @@ func (bc *batchContainer) Add(
 		}
 
 		bc.cmdSend.Send.SequenceId = proto.Uint64(sequenceID)
+		if useTxn {
+			bc.cmdSend.Send.TxnidMostBits = proto.Uint64(mostSigBits)
+			bc.cmdSend.Send.TxnidLeastBits = proto.Uint64(leastSigBits)
+		}
 	}
 	addSingleMessageToBatch(bc.buffer, metadata, payload)
 
diff --git a/pulsar/internal/commands.go b/pulsar/internal/commands.go
index 00e075b..7471ee0 100644
--- a/pulsar/internal/commands.go
+++ b/pulsar/internal/commands.go
@@ -22,11 +22,10 @@ import (
 	"errors"
 	"fmt"
 
-	"google.golang.org/protobuf/proto"
-
 	"github.com/apache/pulsar-client-go/pulsar/internal/compression"
 	"github.com/apache/pulsar-client-go/pulsar/internal/crypto"
 	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
+	"google.golang.org/protobuf/proto"
 )
 
 const (
@@ -332,7 +331,10 @@ func SingleSend(wb Buffer,
 	msgMetadata *pb.MessageMetadata,
 	compressedPayload Buffer,
 	encryptor crypto.Encryptor,
-	maxMassageSize uint32) error {
+	maxMassageSize uint32,
+	useTxn bool,
+	mostSigBits uint64,
+	leastSigBits uint64) error {
 	cmdSend := baseCommand(
 		pb.BaseCommand_SEND,
 		&pb.CommandSend{
@@ -344,6 +346,10 @@ func SingleSend(wb Buffer,
 		isChunk := true
 		cmdSend.Send.IsChunk = &isChunk
 	}
+	if useTxn {
+		cmdSend.Send.TxnidMostBits = proto.Uint64(mostSigBits)
+		cmdSend.Send.TxnidLeastBits = proto.Uint64(leastSigBits)
+	}
 	// payload has been compressed so compressionProvider can be nil
 	return serializeMessage(wb, cmdSend, msgMetadata, compressedPayload,
 		nil, encryptor, maxMassageSize, false)
diff --git a/pulsar/internal/http_client.go b/pulsar/internal/http_client.go
index dccc143..2d440d6 100644
--- a/pulsar/internal/http_client.go
+++ b/pulsar/internal/http_client.go
@@ -173,6 +173,7 @@ func (c *httpClient) GetWithQueryParams(endpoint string, obj interface{}, params
 	return c.GetWithOptions(endpoint, obj, params, decode, nil)
 }
 
+//nolint:bodyclose // false positive
 func (c *httpClient) GetWithOptions(endpoint string, obj interface{}, params map[string]string,
 	decode bool, file io.Writer) ([]byte, error) {
 
@@ -189,7 +190,9 @@ func (c *httpClient) GetWithOptions(endpoint string, obj interface{}, params map
 		req.params = query
 	}
 
-	resp, err := checkSuccessful(c.doRequest(req))
+	doRequest, err := c.doRequest(req)
+	defer safeRespClose(doRequest)
+	resp, err := checkSuccessful(doRequest, err)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pulsar/internal/key_based_batch_builder.go b/pulsar/internal/key_based_batch_builder.go
index 334e674..88a4d5e 100644
--- a/pulsar/internal/key_based_batch_builder.go
+++ b/pulsar/internal/key_based_batch_builder.go
@@ -132,6 +132,9 @@ func (bc *keyBasedBatchContainer) Add(
 	payload []byte,
 	callback interface{}, replicateTo []string, deliverAt time.Time,
 	schemaVersion []byte, multiSchemaEnabled bool,
+	useTxn bool,
+	mostSigBits uint64,
+	leastSigBits uint64,
 ) bool {
 	if replicateTo != nil && bc.numMessages != 0 {
 		// If the current batch is not empty and we're trying to set the replication clusters,
@@ -162,7 +165,7 @@ func (bc *keyBasedBatchContainer) Add(
 	add := batchPart.Add(
 		metadata, sequenceIDGenerator, payload, callback, replicateTo,
 		deliverAt,
-		schemaVersion, multiSchemaEnabled,
+		schemaVersion, multiSchemaEnabled, useTxn, mostSigBits, leastSigBits,
 	)
 	if !add {
 		return false
diff --git a/pulsar/message.go b/pulsar/message.go
index 98190e9..83afd3f 100644
--- a/pulsar/message.go
+++ b/pulsar/message.go
@@ -69,6 +69,10 @@ type ProducerMessage struct {
 	//Schema assign to the current message
 	//Note: messages may have a different schema from producer schema, use it instead of producer schema when assigned
 	Schema Schema
+
+	//Transaction assign to the current message
+	//Note: The message is not visible before the transaction is committed.
+	Transaction Transaction
 }
 
 // Message abstraction used in Pulsar
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index 744df79..e77c929 100644
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -663,6 +663,7 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 					chunkID:          chunkID,
 					uuid:             uuid,
 					chunkRecorder:    cr,
+					transaction:      request.transaction,
 				}
 				// the permit of first chunk has acquired
 				if chunkID != 0 && !p.canAddToQueue(nsr, 0) {
@@ -681,16 +682,16 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 	} else {
 		smm := p.genSingleMessageMetadataInBatch(msg, uncompressedSize)
 		multiSchemaEnabled := !p.options.DisableMultiSchema
-		added := p.batchBuilder.Add(smm, p.sequenceIDGenerator, uncompressedPayload, request,
-			msg.ReplicationClusters, deliverAt, schemaVersion, multiSchemaEnabled)
+		added := addRequestToBatch(smm, p, uncompressedPayload, request, msg, deliverAt, schemaVersion,
+			multiSchemaEnabled)
 		if !added {
-			// The current batch is full.. flush it and retry
+			// The current batch is full. flush it and retry
 
 			p.internalFlushCurrentBatch()
 
 			// after flushing try again to add the current payload
-			if ok := p.batchBuilder.Add(smm, p.sequenceIDGenerator, uncompressedPayload, request,
-				msg.ReplicationClusters, deliverAt, schemaVersion, multiSchemaEnabled); !ok {
+			if ok := addRequestToBatch(smm, p, uncompressedPayload, request, msg, deliverAt, schemaVersion,
+				multiSchemaEnabled); !ok {
 				p.releaseSemaphoreAndMem(uncompressedPayloadSize)
 				request.callback(nil, request.msg, errFailAddToBatch)
 				p.log.WithField("size", uncompressedSize).
@@ -707,6 +708,23 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 	}
 }
 
+func addRequestToBatch(smm *pb.SingleMessageMetadata, p *partitionProducer,
+	uncompressedPayload []byte,
+	request *sendRequest, msg *ProducerMessage, deliverAt time.Time,
+	schemaVersion []byte, multiSchemaEnabled bool) bool {
+	var ok bool
+	if request.transaction != nil {
+		txnID := request.transaction.GetTxnID()
+		ok = p.batchBuilder.Add(smm, p.sequenceIDGenerator, uncompressedPayload, request,
+			msg.ReplicationClusters, deliverAt, schemaVersion, multiSchemaEnabled, true, txnID.MostSigBits,
+			txnID.LeastSigBits)
+	} else {
+		ok = p.batchBuilder.Add(smm, p.sequenceIDGenerator, uncompressedPayload, request,
+			msg.ReplicationClusters, deliverAt, schemaVersion, multiSchemaEnabled, false, 0, 0)
+	}
+	return ok
+}
+
 func (p *partitionProducer) genMetadata(msg *ProducerMessage,
 	uncompressedSize int,
 	deliverAt time.Time) (mm *pb.MessageMetadata) {
@@ -789,16 +807,36 @@ func (p *partitionProducer) internalSingleSend(mm *pb.MessageMetadata,
 	}
 
 	sid := *mm.SequenceId
-
-	if err := internal.SingleSend(
-		buffer,
-		p.producerID,
-		sid,
-		mm,
-		payloadBuf,
-		p.encryptor,
-		maxMessageSize,
-	); err != nil {
+	var err error
+	if request.transaction != nil {
+		txnID := request.transaction.GetTxnID()
+		err = internal.SingleSend(
+			buffer,
+			p.producerID,
+			sid,
+			mm,
+			payloadBuf,
+			p.encryptor,
+			maxMessageSize,
+			true,
+			txnID.MostSigBits,
+			txnID.LeastSigBits,
+		)
+	} else {
+		err = internal.SingleSend(
+			buffer,
+			p.producerID,
+			sid,
+			mm,
+			payloadBuf,
+			p.encryptor,
+			maxMessageSize,
+			false,
+			0,
+			0,
+		)
+	}
+	if err != nil {
 		request.callback(nil, request.msg, err)
 		p.releaseSemaphoreAndMem(int64(len(msg.Payload)))
 		p.log.WithError(err).Errorf("Single message serialize failed %s", msg.Value)
@@ -952,6 +990,9 @@ func (p *partitionProducer) failTimeoutMessages() {
 						sr.callback(nil, sr.msg, errSendTimeout)
 					})
 				}
+				if sr.transaction != nil {
+					sr.transaction.endSendOrAckOp(nil)
+				}
 			}
 
 			// flag the send has completed with error, flush make no effect
@@ -1067,9 +1108,34 @@ func (p *partitionProducer) SendAsync(ctx context.Context, msg *ProducerMessage,
 
 func (p *partitionProducer) internalSendAsync(ctx context.Context, msg *ProducerMessage,
 	callback func(MessageID, *ProducerMessage, error), flushImmediately bool) {
+	//Register transaction operation to transaction and the transaction coordinator.
+	var newCallback func(MessageID, *ProducerMessage, error)
+	if msg.Transaction != nil {
+		transactionImpl := (msg.Transaction).(*transaction)
+		if transactionImpl.state != TxnOpen {
+			p.log.WithField("state", transactionImpl.state).Error("Failed to send message" +
+				" by a non-open transaction.")
+			callback(nil, msg, newError(InvalidStatus, "Failed to send message by a non-open transaction."))
+			return
+		}
+
+		if err := transactionImpl.registerProducerTopic(p.topic); err != nil {
+			callback(nil, msg, err)
+			return
+		}
+		if err := transactionImpl.registerSendOrAckOp(); err != nil {
+			callback(nil, msg, err)
+		}
+		newCallback = func(id MessageID, producerMessage *ProducerMessage, err error) {
+			callback(id, producerMessage, err)
+			transactionImpl.endSendOrAckOp(err)
+		}
+	} else {
+		newCallback = callback
+	}
 	if p.getProducerState() != producerReady {
 		// Producer is closing
-		callback(nil, msg, errProducerClosed)
+		newCallback(nil, msg, errProducerClosed)
 		return
 	}
 
@@ -1078,16 +1144,20 @@ func (p *partitionProducer) internalSendAsync(ctx context.Context, msg *Producer
 
 	// callbackOnce make sure the callback is only invoked once in chunking
 	callbackOnce := &sync.Once{}
-
+	var txn *transaction
+	if msg.Transaction != nil {
+		txn = (msg.Transaction).(*transaction)
+	}
 	sr := &sendRequest{
 		ctx:              ctx,
 		msg:              msg,
-		callback:         callback,
+		callback:         newCallback,
 		callbackOnce:     callbackOnce,
 		flushImmediately: flushImmediately,
 		publishTime:      time.Now(),
 		blockCh:          bc,
 		closeBlockChOnce: &sync.Once{},
+		transaction:      txn,
 	}
 	p.options.Interceptors.BeforeSend(p, msg)
 
@@ -1191,6 +1261,9 @@ func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt)
 					p.options.Interceptors.OnSendAcknowledgement(p, sr.msg, msgID)
 				}
 			}
+			if sr.transaction != nil {
+				sr.transaction.endSendOrAckOp(nil)
+			}
 		}
 
 		// Mark this pending item as done
@@ -1287,6 +1360,7 @@ type sendRequest struct {
 	chunkID          int
 	uuid             string
 	chunkRecorder    *chunkRecorder
+	transaction      *transaction
 }
 
 // stopBlock can be invoked multiple times safety
diff --git a/pulsar/schema.go b/pulsar/schema.go
index 997d85f..405049d 100644
--- a/pulsar/schema.go
+++ b/pulsar/schema.go
@@ -30,6 +30,9 @@ import (
 
 	"github.com/linkedin/goavro/v2"
 	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/reflect/protodesc"
+	"google.golang.org/protobuf/reflect/protoreflect"
+	"google.golang.org/protobuf/types/descriptorpb"
 )
 
 type SchemaType int
@@ -55,6 +58,7 @@ const (
 	AUTO        = -2              //
 	AutoConsume = -3              //Auto Consume Type.
 	AutoPublish = -4              // Auto Publish Type.
+	ProtoNative = 20              //Protobuf native message encoding and decoding
 )
 
 // Encapsulates data around the schema definition
@@ -102,6 +106,8 @@ func NewSchema(schemaType SchemaType, schemaData []byte, properties map[string]s
 		s = NewFloatSchema(properties)
 	case DOUBLE:
 		s = NewDoubleSchema(properties)
+	case ProtoNative:
+		s = newProtoNativeSchema(schemaDef, properties)
 	default:
 		err = fmt.Errorf("not support schema type of %v", schemaType)
 	}
@@ -219,6 +225,84 @@ func (ps *ProtoSchema) GetSchemaInfo() *SchemaInfo {
 	return &ps.SchemaInfo
 }
 
+type ProtoNativeSchema struct {
+	SchemaInfo
+}
+
+func NewProtoNativeSchemaWithMessage(message proto.Message, properties map[string]string) *ProtoNativeSchema {
+	schemaDef, err := getProtoNativeSchemaInfo(message)
+	if err != nil {
+		log.Fatalf("Get ProtoNative schema info error:%v", err)
+	}
+	return newProtoNativeSchema(schemaDef, properties)
+}
+
+func newProtoNativeSchema(protoNativeSchemaDef string, properties map[string]string) *ProtoNativeSchema {
+	pns := new(ProtoNativeSchema)
+	pns.SchemaInfo.Schema = protoNativeSchemaDef
+	pns.SchemaInfo.Type = ProtoNative
+	pns.SchemaInfo.Properties = properties
+	pns.SchemaInfo.Name = "ProtoNative"
+	return pns
+}
+
+func getProtoNativeSchemaInfo(message proto.Message) (string, error) {
+	fileDesc := message.ProtoReflect().Descriptor().ParentFile()
+	fileProtoMap := make(map[string]*descriptorpb.FileDescriptorProto)
+	getFileProto(fileDesc, fileProtoMap)
+
+	fileDescList := make([]*descriptorpb.FileDescriptorProto, 0, len(fileProtoMap))
+	for _, v := range fileProtoMap {
+		fileDescList = append(fileDescList, v)
+	}
+	fileDescSet := descriptorpb.FileDescriptorSet{
+		File: fileDescList,
+	}
+	bytesData, err := proto.Marshal(&fileDescSet)
+	if err != nil {
+		return "", err
+	}
+	schemaData := ProtoNativeSchemaData{
+		FileDescriptorSet:      bytesData,
+		RootMessageTypeName:    string(message.ProtoReflect().Descriptor().FullName()),
+		RootFileDescriptorName: fileDesc.Path(),
+	}
+	jsonData, err := json.Marshal(schemaData)
+	if err != nil {
+		return "", err
+	}
+	return string(jsonData), nil
+}
+
+type ProtoNativeSchemaData struct {
+	FileDescriptorSet      []byte `json:"fileDescriptorSet"`
+	RootMessageTypeName    string `json:"rootMessageTypeName"`
+	RootFileDescriptorName string `json:"rootFileDescriptorName"`
+}
+
+func getFileProto(fileDesc protoreflect.FileDescriptor, protoMap map[string]*descriptorpb.FileDescriptorProto) {
+	for i := 0; i < fileDesc.Imports().Len(); i++ {
+		getFileProto(fileDesc.Imports().Get(i).ParentFile(), protoMap)
+	}
+	protoMap[fileDesc.Path()] = protodesc.ToFileDescriptorProto(fileDesc)
+}
+
+func (ps *ProtoNativeSchema) Encode(data interface{}) ([]byte, error) {
+	return proto.Marshal(data.(proto.Message))
+}
+
+func (ps *ProtoNativeSchema) Decode(data []byte, v interface{}) error {
+	return proto.Unmarshal(data, v.(proto.Message))
+}
+
+func (ps *ProtoNativeSchema) Validate(message []byte) error {
+	return ps.Decode(message, nil)
+}
+
+func (ps *ProtoNativeSchema) GetSchemaInfo() *SchemaInfo {
+	return &ps.SchemaInfo
+}
+
 type AvroSchema struct {
 	AvroCodec
 	SchemaInfo
diff --git a/pulsar/transaction.go b/pulsar/transaction.go
index 60e1d2b..944c7e3 100644
--- a/pulsar/transaction.go
+++ b/pulsar/transaction.go
@@ -49,10 +49,10 @@ const (
 
 // TxnID An identifier for representing a transaction.
 type TxnID struct {
-	// mostSigBits The most significant 64 bits of this TxnID.
-	mostSigBits uint64
-	// leastSigBits The least significant 64 bits of this TxnID.
-	leastSigBits uint64
+	// MostSigBits The most significant 64 bits of this TxnID.
+	MostSigBits uint64
+	// LeastSigBits The least significant 64 bits of this TxnID.
+	LeastSigBits uint64
 }
 
 // Transaction used to guarantee exactly-once
diff --git a/pulsar/transaction_coordinator_client.go b/pulsar/transaction_coordinator_client.go
index 1535fad..96cca87 100644
--- a/pulsar/transaction_coordinator_client.go
+++ b/pulsar/transaction_coordinator_client.go
@@ -135,11 +135,11 @@ func (tc *transactionCoordinatorClient) addPublishPartitionToTxn(id *TxnID, part
 	requestID := tc.client.rpcClient.NewRequestID()
 	cmdAddPartitions := &pb.CommandAddPartitionToTxn{
 		RequestId:      proto.Uint64(requestID),
-		TxnidMostBits:  proto.Uint64(id.mostSigBits),
-		TxnidLeastBits: proto.Uint64(id.leastSigBits),
+		TxnidMostBits:  proto.Uint64(id.MostSigBits),
+		TxnidLeastBits: proto.Uint64(id.LeastSigBits),
 		Partitions:     partitions,
 	}
-	res, err := tc.client.rpcClient.RequestOnCnx(tc.cons[id.mostSigBits], requestID,
+	res, err := tc.client.rpcClient.RequestOnCnx(tc.cons[id.MostSigBits], requestID,
 		pb.BaseCommand_ADD_PARTITION_TO_TXN, cmdAddPartitions)
 	tc.semaphore.Release()
 	if err != nil {
@@ -163,11 +163,11 @@ func (tc *transactionCoordinatorClient) addSubscriptionToTxn(id *TxnID, topic st
 	}
 	cmdAddSubscription := &pb.CommandAddSubscriptionToTxn{
 		RequestId:      proto.Uint64(requestID),
-		TxnidMostBits:  proto.Uint64(id.mostSigBits),
-		TxnidLeastBits: proto.Uint64(id.leastSigBits),
+		TxnidMostBits:  proto.Uint64(id.MostSigBits),
+		TxnidLeastBits: proto.Uint64(id.LeastSigBits),
 		Subscription:   []*pb.Subscription{sub},
 	}
-	res, err := tc.client.rpcClient.RequestOnCnx(tc.cons[id.mostSigBits], requestID,
+	res, err := tc.client.rpcClient.RequestOnCnx(tc.cons[id.MostSigBits], requestID,
 		pb.BaseCommand_ADD_SUBSCRIPTION_TO_TXN, cmdAddSubscription)
 	tc.semaphore.Release()
 	if err != nil {
@@ -187,10 +187,10 @@ func (tc *transactionCoordinatorClient) endTxn(id *TxnID, action pb.TxnAction) e
 	cmdEndTxn := &pb.CommandEndTxn{
 		RequestId:      proto.Uint64(requestID),
 		TxnAction:      &action,
-		TxnidMostBits:  proto.Uint64(id.mostSigBits),
-		TxnidLeastBits: proto.Uint64(id.leastSigBits),
+		TxnidMostBits:  proto.Uint64(id.MostSigBits),
+		TxnidLeastBits: proto.Uint64(id.LeastSigBits),
 	}
-	res, err := tc.client.rpcClient.RequestOnCnx(tc.cons[id.mostSigBits], requestID, pb.BaseCommand_END_TXN, cmdEndTxn)
+	res, err := tc.client.rpcClient.RequestOnCnx(tc.cons[id.MostSigBits], requestID, pb.BaseCommand_END_TXN, cmdEndTxn)
 	tc.semaphore.Release()
 	if err != nil {
 		return err
diff --git a/pulsar/transaction_impl.go b/pulsar/transaction_impl.go
index 7cc93ec..8a24c46 100644
--- a/pulsar/transaction_impl.go
+++ b/pulsar/transaction_impl.go
@@ -231,6 +231,8 @@ func (state TxnState) string() string {
 		return "TxnAborted"
 	case TxnTimeout:
 		return "TxnTimeout"
+	case TxnError:
+		return "TxnError"
 	default:
 		return "Unknown"
 	}


Test patch:
diff --git a/integration-tests/pb/build.sh b/integration-tests/pb/build.sh
index 32c0b08..e192458 100755
--- a/integration-tests/pb/build.sh
+++ b/integration-tests/pb/build.sh
@@ -18,4 +18,9 @@
 # under the License.
 #
 
-protoc --go_out=. --go_opt=paths=source_relative hello.proto
+PROJECT_DIR=$(git rev-parse --show-toplevel)
+SOURCE_PATH="$PROJECT_DIR/integration-tests/pb"
+DEST_PATH="$PROJECT_DIR/integration-tests/pb"
+echo "source_path: $SOURCE_PATH"
+echo "dest_path: $DEST_PATH"
+protoc -I=$SOURCE_PATH --go_out=$DEST_PATH $SOURCE_PATH/*.proto
diff --git a/integration-tests/pb/hello.pb.go b/integration-tests/pb/hello.pb.go
index 7884dfa..541172c 100644
--- a/integration-tests/pb/hello.pb.go
+++ b/integration-tests/pb/hello.pb.go
@@ -18,11 +18,11 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.28.1
-// 	protoc        v3.21.9
+// 	protoc-gen-go v1.30.0
+// 	protoc        v3.21.12
 // source: hello.proto
 
-package pb
+package __
 
 import (
 	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
@@ -43,8 +43,10 @@ type Test struct {
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
-	Num int32  `protobuf:"varint,1,opt,name=num,proto3" json:"num,omitempty"`
-	Msf string `protobuf:"bytes,2,opt,name=msf,proto3" json:"msf,omitempty"`
+	Num       int32      `protobuf:"varint,1,opt,name=num,proto3" json:"num,omitempty"`
+	Msf       string     `protobuf:"bytes,2,opt,name=msf,proto3" json:"msf,omitempty"`
+	Foo       *Foo       `protobuf:"bytes,3,opt,name=foo,proto3" json:"foo,omitempty"`
+	HiContent *HiContent `protobuf:"bytes,4,opt,name=hiContent,proto3" json:"hiContent,omitempty"`
 }
 
 func (x *Test) Reset() {
@@ -93,15 +95,102 @@ func (x *Test) GetMsf() string {
 	return ""
 }
 
+func (x *Test) GetFoo() *Foo {
+	if x != nil {
+		return x.Foo
+	}
+	return nil
+}
+
+func (x *Test) GetHiContent() *HiContent {
+	if x != nil {
+		return x.HiContent
+	}
+	return nil
+}
+
+type Foo struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Name  string   `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
+	Value int32    `protobuf:"varint,2,opt,name=value,proto3" json:"value,omitempty"`
+	Names []string `protobuf:"bytes,3,rep,name=names,proto3" json:"names,omitempty"`
+}
+
+func (x *Foo) Reset() {
+	*x = Foo{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_hello_proto_msgTypes[1]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *Foo) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*Foo) ProtoMessage() {}
+
+func (x *Foo) ProtoReflect() protoreflect.Message {
+	mi := &file_hello_proto_msgTypes[1]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use Foo.ProtoReflect.Descriptor instead.
+func (*Foo) Descriptor() ([]byte, []int) {
+	return file_hello_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *Foo) GetName() string {
+	if x != nil {
+		return x.Name
+	}
+	return ""
+}
+
+func (x *Foo) GetValue() int32 {
+	if x != nil {
+		return x.Value
+	}
+	return 0
+}
+
+func (x *Foo) GetNames() []string {
+	if x != nil {
+		return x.Names
+	}
+	return nil
+}
+
 var File_hello_proto protoreflect.FileDescriptor
 
 var file_hello_proto_rawDesc = []byte{
 	0x0a, 0x0b, 0x68, 0x65, 0x6c, 0x6c, 0x6f, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x09, 0x70,
-	0x72, 0x6f, 0x74, 0x6f, 0x74, 0x65, 0x73, 0x74, 0x22, 0x2a, 0x0a, 0x04, 0x54, 0x65, 0x73, 0x74,
-	0x12, 0x10, 0x0a, 0x03, 0x6e, 0x75, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x05, 0x52, 0x03, 0x6e,
-	0x75, 0x6d, 0x12, 0x10, 0x0a, 0x03, 0x6d, 0x73, 0x66, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52,
-	0x03, 0x6d, 0x73, 0x66, 0x42, 0x06, 0x5a, 0x04, 0x2e, 0x2f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72,
-	0x6f, 0x74, 0x6f, 0x33,
+	0x72, 0x6f, 0x74, 0x6f, 0x74, 0x65, 0x73, 0x74, 0x1a, 0x08, 0x68, 0x69, 0x2e, 0x70, 0x72, 0x6f,
+	0x74, 0x6f, 0x22, 0x80, 0x01, 0x0a, 0x04, 0x54, 0x65, 0x73, 0x74, 0x12, 0x10, 0x0a, 0x03, 0x6e,
+	0x75, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x05, 0x52, 0x03, 0x6e, 0x75, 0x6d, 0x12, 0x10, 0x0a,
+	0x03, 0x6d, 0x73, 0x66, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6d, 0x73, 0x66, 0x12,
+	0x20, 0x0a, 0x03, 0x66, 0x6f, 0x6f, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x74, 0x65, 0x73, 0x74, 0x2e, 0x46, 0x6f, 0x6f, 0x52, 0x03, 0x66, 0x6f,
+	0x6f, 0x12, 0x32, 0x0a, 0x09, 0x68, 0x69, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x18, 0x04,
+	0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x74, 0x65, 0x73, 0x74,
+	0x2e, 0x48, 0x69, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x52, 0x09, 0x68, 0x69, 0x43, 0x6f,
+	0x6e, 0x74, 0x65, 0x6e, 0x74, 0x22, 0x45, 0x0a, 0x03, 0x46, 0x6f, 0x6f, 0x12, 0x12, 0x0a, 0x04,
+	0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65,
+	0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52,
+	0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x18,
+	0x03, 0x20, 0x03, 0x28, 0x09, 0x52, 0x05, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x42, 0x03, 0x5a, 0x01,
+	0x2e, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -116,16 +205,20 @@ func file_hello_proto_rawDescGZIP() []byte {
 	return file_hello_proto_rawDescData
 }
 
-var file_hello_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
+var file_hello_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
 var file_hello_proto_goTypes = []interface{}{
-	(*Test)(nil), // 0: prototest.Test
+	(*Test)(nil),      // 0: prototest.Test
+	(*Foo)(nil),       // 1: prototest.Foo
+	(*HiContent)(nil), // 2: prototest.HiContent
 }
 var file_hello_proto_depIdxs = []int32{
-	0, // [0:0] is the sub-list for method output_type
-	0, // [0:0] is the sub-list for method input_type
-	0, // [0:0] is the sub-list for extension type_name
-	0, // [0:0] is the sub-list for extension extendee
-	0, // [0:0] is the sub-list for field type_name
+	1, // 0: prototest.Test.foo:type_name -> prototest.Foo
+	2, // 1: prototest.Test.hiContent:type_name -> prototest.HiContent
+	2, // [2:2] is the sub-list for method output_type
+	2, // [2:2] is the sub-list for method input_type
+	2, // [2:2] is the sub-list for extension type_name
+	2, // [2:2] is the sub-list for extension extendee
+	0, // [0:2] is the sub-list for field type_name
 }
 
 func init() { file_hello_proto_init() }
@@ -133,6 +226,7 @@ func file_hello_proto_init() {
 	if File_hello_proto != nil {
 		return
 	}
+	file_hi_proto_init()
 	if !protoimpl.UnsafeEnabled {
 		file_hello_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
 			switch v := v.(*Test); i {
@@ -146,6 +240,18 @@ func file_hello_proto_init() {
 				return nil
 			}
 		}
+		file_hello_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*Foo); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
 	}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
@@ -153,7 +259,7 @@ func file_hello_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_hello_proto_rawDesc,
 			NumEnums:      0,
-			NumMessages:   1,
+			NumMessages:   2,
 			NumExtensions: 0,
 			NumServices:   0,
 		},
diff --git a/integration-tests/pb/hello.proto b/integration-tests/pb/hello.proto
index 3a67c3b..ca6a4ff 100644
--- a/integration-tests/pb/hello.proto
+++ b/integration-tests/pb/hello.proto
@@ -19,9 +19,19 @@
 syntax = "proto3";
 package prototest;
 
-option go_package = "./pb";
+option go_package = ".";
+
+import "hi.proto";
 
 message Test {
     int32 num = 1;
     string msf = 2;
+    Foo foo = 3;
+    HiContent hiContent = 4;
+}
+
+message Foo {
+    string name = 1;
+    int32 value = 2;
+    repeated string names = 3;
 }
diff --git a/integration-tests/pb/hi.pb.go b/integration-tests/pb/hi.pb.go
new file mode 100644
index 0000000..1027caf
--- /dev/null
+++ b/integration-tests/pb/hi.pb.go
@@ -0,0 +1,168 @@
+//*
+// 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.
+
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.30.0
+// 	protoc        v3.21.12
+// source: hi.proto
+
+package __
+
+import (
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+type HiContent struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Id      int32  `protobuf:"varint,2,opt,name=id,proto3" json:"id,omitempty"`
+	Content string `protobuf:"bytes,1,opt,name=content,proto3" json:"content,omitempty"`
+}
+
+func (x *HiContent) Reset() {
+	*x = HiContent{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_hi_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *HiContent) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*HiContent) ProtoMessage() {}
+
+func (x *HiContent) ProtoReflect() protoreflect.Message {
+	mi := &file_hi_proto_msgTypes[0]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use HiContent.ProtoReflect.Descriptor instead.
+func (*HiContent) Descriptor() ([]byte, []int) {
+	return file_hi_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *HiContent) GetId() int32 {
+	if x != nil {
+		return x.Id
+	}
+	return 0
+}
+
+func (x *HiContent) GetContent() string {
+	if x != nil {
+		return x.Content
+	}
+	return ""
+}
+
+var File_hi_proto protoreflect.FileDescriptor
+
+var file_hi_proto_rawDesc = []byte{
+	0x0a, 0x08, 0x68, 0x69, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x09, 0x70, 0x72, 0x6f, 0x74,
+	0x6f, 0x74, 0x65, 0x73, 0x74, 0x22, 0x35, 0x0a, 0x09, 0x48, 0x69, 0x43, 0x6f, 0x6e, 0x74, 0x65,
+	0x6e, 0x74, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x02,
+	0x69, 0x64, 0x12, 0x18, 0x0a, 0x07, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x18, 0x01, 0x20,
+	0x01, 0x28, 0x09, 0x52, 0x07, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x42, 0x03, 0x5a, 0x01,
+	0x2e, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+}
+
+var (
+	file_hi_proto_rawDescOnce sync.Once
+	file_hi_proto_rawDescData = file_hi_proto_rawDesc
+)
+
+func file_hi_proto_rawDescGZIP() []byte {
+	file_hi_proto_rawDescOnce.Do(func() {
+		file_hi_proto_rawDescData = protoimpl.X.CompressGZIP(file_hi_proto_rawDescData)
+	})
+	return file_hi_proto_rawDescData
+}
+
+var file_hi_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
+var file_hi_proto_goTypes = []interface{}{
+	(*HiContent)(nil), // 0: prototest.HiContent
+}
+var file_hi_proto_depIdxs = []int32{
+	0, // [0:0] is the sub-list for method output_type
+	0, // [0:0] is the sub-list for method input_type
+	0, // [0:0] is the sub-list for extension type_name
+	0, // [0:0] is the sub-list for extension extendee
+	0, // [0:0] is the sub-list for field type_name
+}
+
+func init() { file_hi_proto_init() }
+func file_hi_proto_init() {
+	if File_hi_proto != nil {
+		return
+	}
+	if !protoimpl.UnsafeEnabled {
+		file_hi_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*HiContent); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: file_hi_proto_rawDesc,
+			NumEnums:      0,
+			NumMessages:   1,
+			NumExtensions: 0,
+			NumServices:   0,
+		},
+		GoTypes:           file_hi_proto_goTypes,
+		DependencyIndexes: file_hi_proto_depIdxs,
+		MessageInfos:      file_hi_proto_msgTypes,
+	}.Build()
+	File_hi_proto = out.File
+	file_hi_proto_rawDesc = nil
+	file_hi_proto_goTypes = nil
+	file_hi_proto_depIdxs = nil
+}
diff --git a/integration-tests/pb/hi.proto b/integration-tests/pb/hi.proto
new file mode 100644
index 0000000..f5a401a
--- /dev/null
+++ b/integration-tests/pb/hi.proto
@@ -0,0 +1,9 @@
+syntax = "proto3";
+package prototest;
+
+option go_package = ".";
+
+message HiContent {
+  int32 id = 2;
+  string content = 1;
+}
diff --git a/pulsar/helper_for_test.go b/pulsar/helper_for_test.go
index 7bbf66e..426855b 100644
--- a/pulsar/helper_for_test.go
+++ b/pulsar/helper_for_test.go
@@ -159,7 +159,7 @@ func topicStats(topic string) (map[string]interface{}, error) {
 
 func transactionStats(id *TxnID) (map[string]interface{}, error) {
 	var metadata map[string]interface{}
-	path := fmt.Sprintf("admin/v3/transactions/transactionMetadata/%d/%d", id.mostSigBits, id.leastSigBits)
+	path := fmt.Sprintf("admin/v3/transactions/transactionMetadata/%d/%d", id.MostSigBits, id.LeastSigBits)
 	err := httpGet(path, &metadata)
 	return metadata, err
 }
diff --git a/pulsar/internal/pulsartracing/consumer_interceptor_test.go b/pulsar/internal/pulsartracing/consumer_interceptor_test.go
index 34e09d5..06c9a58 100644
--- a/pulsar/internal/pulsartracing/consumer_interceptor_test.go
+++ b/pulsar/internal/pulsartracing/consumer_interceptor_test.go
@@ -52,6 +52,10 @@ func (c *mockConsumer) Subscription() string {
 	return ""
 }
 
+func (c *mockConsumer) AckWithTxn(msg pulsar.Message, txn pulsar.Transaction) error {
+	return nil
+}
+
 func (c *mockConsumer) Unsubscribe() error {
 	return nil
 }
diff --git a/pulsar/schema_test.go b/pulsar/schema_test.go
index e7d52ca..c2008f6 100644
--- a/pulsar/schema_test.go
+++ b/pulsar/schema_test.go
@@ -22,7 +22,7 @@ import (
 	"log"
 	"testing"
 
-	"github.com/apache/pulsar-client-go/integration-tests/pb"
+	pb "github.com/apache/pulsar-client-go/integration-tests/pb"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -147,6 +147,69 @@ func TestProtoSchema(t *testing.T) {
 	defer consumer.Close()
 }
 
+func TestProtoNativeSchema(t *testing.T) {
+	client := createClient()
+	defer client.Close()
+
+	topic := "proto-native"
+
+	// create producer
+	psProducer := NewProtoNativeSchemaWithMessage(&pb.Test{}, nil)
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:  topic,
+		Schema: psProducer,
+	})
+	assert.Nil(t, err)
+
+	names := []string{
+		"name-a",
+		"name-b",
+		"name-c",
+	}
+	if _, err := producer.Send(context.Background(), &ProducerMessage{
+		Value: &pb.Test{
+			Num: 100,
+			Msf: "pulsar",
+			Foo: &pb.Foo{
+				Name:  "foo name",
+				Value: 200,
+				Names: names,
+			},
+			HiContent: &pb.HiContent{
+				Id:      300,
+				Content: "hi content",
+			},
+		},
+	}); err != nil {
+		log.Fatal(err)
+	}
+
+	//create consumer
+	unobj := pb.Test{}
+	psConsumer := NewProtoNativeSchemaWithMessage(&pb.Test{}, nil)
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:                       topic,
+		SubscriptionName:            "sub-1",
+		Schema:                      psConsumer,
+		SubscriptionInitialPosition: SubscriptionPositionEarliest,
+	})
+	assert.Nil(t, err)
+
+	msg, err := consumer.Receive(context.Background())
+	assert.Nil(t, err)
+	err = msg.GetSchemaValue(&unobj)
+	assert.Nil(t, err)
+	assert.Equal(t, unobj.Num, int32(100))
+	assert.Equal(t, unobj.Msf, "pulsar")
+	assert.Equal(t, unobj.Foo.Name, "foo name")
+	assert.Equal(t, unobj.Foo.Value, int32(200))
+	assert.Equal(t, unobj.Foo.Names, names)
+	assert.Equal(t, unobj.HiContent.Id, int32(300))
+	assert.Equal(t, unobj.HiContent.Content, "hi content")
+	consumer.Ack(msg)
+	defer consumer.Close()
+}
+
 func TestAvroSchema(t *testing.T) {
 	client := createClient()
 	defer client.Close()
diff --git a/pulsar/transaction_test.go b/pulsar/transaction_test.go
index 362e4d2..66a82cc 100644
--- a/pulsar/transaction_test.go
+++ b/pulsar/transaction_test.go
@@ -103,8 +103,8 @@ func TestTxnImplCommitOrAbort(t *testing.T) {
 	//The operations of committing txn1 should success at the first time and fail at the second time.
 	txn1 := createTxn(tc, 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))
+	require.Nil(t, err, fmt.Sprintf("Failed to commit the transaction %d:%d\n", txn1.txnID.MostSigBits,
+		txn1.txnID.LeastSigBits))
 	txn1.state = TxnOpen
 	txn1.opsFlow <- true
 	err = txn1.Commit(context.Background())
@@ -117,7 +117,7 @@ func TestTxnImplCommitOrAbort(t *testing.T) {
 	txn2 := newTransaction(*id2, tc, time.Hour)
 	err = txn2.Abort(context.Background())
 	require.Nil(t, err, fmt.Sprintf("Failed to abort the transaction %d:%d\n",
-		id2.mostSigBits, id2.leastSigBits))
+		id2.MostSigBits, id2.LeastSigBits))
 	txn2.state = TxnOpen
 	txn2.opsFlow <- true
 	err = txn2.Abort(context.Background())
@@ -209,11 +209,237 @@ func createTcClient(t *testing.T) (*transactionCoordinatorClient, *client) {
 		URL:                   webServiceURLTLS,
 		TLSTrustCertsFilePath: caCertsPath,
 		Authentication:        NewAuthenticationTLS(tlsClientCertPath, tlsClientKeyPath),
+		EnableTransaction:     true,
 	})
 	require.Nil(t, err, "Failed to create client.")
-	tcClient := newTransactionCoordinatorClientImpl(c.(*client))
-	err = tcClient.start()
-	require.Nil(t, err, "Failed to start transaction coordinator.")
 
-	return tcClient, c.(*client)
+	return c.(*client).tcClient, c.(*client)
+}
+
+// TestConsumeAndProduceWithTxn is a test function that validates 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.
+// 2. Prepare: Create a topic and a subscription.
+// 3. Produce 10 messages with a transaction and 10 messages without a transaction.
+// - Expectation: The consumer should be able to receive the 10 messages sent without a transaction,
+// but not the 10 messages sent with the transaction.
+// 4. Commit the transaction and receive the remaining 10 messages.
+// - Expectation: The consumer should be able to receive the 10 messages sent with the transaction.
+// 5. Clean up: Close the consumer and producer instances.
+//
+// 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) {
+	// Step 1: Prepare - Create PulsarClient and initialize the transaction coordinator client.
+	topic := newTopicName()
+	sub := "my-sub"
+	_, client := createTcClient(t)
+	// Step 2: Prepare - Create Topic and Subscription.
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:            topic,
+		SubscriptionName: sub,
+	})
+	assert.NoError(t, err)
+	producer, _ := client.CreateProducer(ProducerOptions{
+		Topic:       topic,
+		SendTimeout: 0,
+	})
+	// 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)
+	require.Nil(t, err)
+	for i := 0; i < 10; i++ {
+		_, err = producer.Send(context.Background(), &ProducerMessage{
+			Payload: make([]byte, 1024),
+		})
+		require.Nil(t, err)
+	}
+	for i := 0; i < 10; i++ {
+		_, err := producer.Send(context.Background(), &ProducerMessage{
+			Transaction: txn,
+			Payload:     make([]byte, 1024),
+		})
+		require.Nil(t, err)
+	}
+	// Attempt to receive and acknowledge the 10 messages sent without a transaction.
+	for i := 0; i < 10; i++ {
+		msg, _ := consumer.Receive(context.Background())
+		assert.NotNil(t, msg)
+		err = consumer.Ack(msg)
+		assert.Nil(t, err)
+	}
+	// Create a goroutine to attempt receiving a message and send it to the 'done1' channel.
+	done1 := make(chan Message)
+	go func() {
+		msg, _ := consumer.Receive(context.Background())
+		err := consumer.AckID(msg.ID())
+		require.Nil(t, err)
+		close(done1)
+	}()
+	// Expectation: The consumer should not receive uncommitted messages.
+	select {
+	case <-done1:
+		require.Fail(t, "The consumer should not receive uncommitted message")
+	case <-time.After(time.Second):
+	}
+	// Step 4: After committing the transaction, we should be able to receive the remaining 10 messages.
+	// 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)
+	require.Nil(t, err)
+	for i := 0; i < 9; i++ {
+		msg, _ := consumer.Receive(context.Background())
+		require.NotNil(t, msg)
+		err = consumer.AckWithTxn(msg, txn)
+		require.Nil(t, err)
+	}
+	_ = txn.Commit(context.Background())
+	<-done1
+	consumer.Close()
+	consumer, _ = client.Subscribe(ConsumerOptions{
+		Topic:            topic,
+		SubscriptionName: sub,
+	})
+	// 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())
+		close(done2)
+	}()
+
+	// Expectation: The consumer should not receive uncommitted messages.
+	select {
+	case <-done2:
+		require.Fail(t, "The consumer should not receive messages")
+	case <-time.After(time.Second):
+	}
+
+	// Step 5: Clean up - Close the consumer and producer instances.
+	consumer.Close()
+	producer.Close()
+}
+
+func TestAckAndSendWithTxn(t *testing.T) {
+	// Prepare: Create PulsarClient and initialize the transaction coordinator client.
+	sourceTopic := newTopicName()
+	sinkTopic := newTopicName()
+	sub := "my-sub"
+	_, client := createTcClient(t)
+
+	// Prepare: Create source and sink topics and subscriptions.
+	sourceConsumer, _ := client.Subscribe(ConsumerOptions{
+		Topic:            sourceTopic,
+		SubscriptionName: sub,
+	})
+	sourceProducer, _ := client.CreateProducer(ProducerOptions{
+		Topic:       sourceTopic,
+		SendTimeout: 0,
+	})
+	sinkConsumer, _ := client.Subscribe(ConsumerOptions{
+		Topic:            sinkTopic,
+		SubscriptionName: sub,
+	})
+	sinkProducer, _ := client.CreateProducer(ProducerOptions{
+		Topic:       sinkTopic,
+		SendTimeout: 0,
+	})
+
+	// Produce 10 messages to the source topic.
+	for i := 0; i < 10; i++ {
+		_, err := sourceProducer.Send(context.Background(), &ProducerMessage{
+			Payload: make([]byte, 1024),
+		})
+		require.Nil(t, err)
+	}
+
+	// Open a transaction and consume messages from the source topic while sending messages to the sink topic.
+	txn, err := client.NewTransaction(time.Hour)
+	require.Nil(t, err)
+
+	for i := 0; i < 10; i++ {
+		msg, _ := sourceConsumer.Receive(context.Background())
+		require.NotNil(t, msg)
+
+		payload := msg.Payload()
+		_, err := sinkProducer.Send(context.Background(), &ProducerMessage{
+			Transaction: txn,
+			Payload:     payload,
+		})
+		require.Nil(t, err)
+
+		err = sourceConsumer.AckWithTxn(msg, txn)
+		require.Nil(t, err)
+	}
+
+	// Commit the transaction.
+	_ = txn.Commit(context.Background())
+
+	// Verify that the messages are available in the sink topic.
+	for i := 0; i < 10; i++ {
+		msg, _ := sinkConsumer.Receive(context.Background())
+		require.NotNil(t, msg)
+		err = sinkConsumer.Ack(msg)
+		require.Nil(t, err)
+	}
+
+	// Clean up: Close consumers and producers.
+	sourceConsumer.Close()
+	sourceProducer.Close()
+	sinkConsumer.Close()
+	sinkProducer.Close()
+}
+
+func TestTransactionAbort(t *testing.T) {
+	// Prepare: Create PulsarClient and initialize the transaction coordinator client.
+	topic := newTopicName()
+	sub := "my-sub"
+	_, client := createTcClient(t)
+
+	// Prepare: Create Topic and Subscription.
+	consumer, _ := client.Subscribe(ConsumerOptions{
+		Topic:            topic,
+		SubscriptionName: sub,
+	})
+	producer, _ := client.CreateProducer(ProducerOptions{
+		Topic:       topic,
+		SendTimeout: 0,
+	})
+
+	// Open a transaction and send 10 messages with the transaction.
+	txn, err := client.NewTransaction(time.Hour)
+	require.Nil(t, err)
+
+	for i := 0; i < 10; i++ {
+		_, err = producer.Send(context.Background(), &ProducerMessage{
+			Transaction: txn,
+			Payload:     make([]byte, 1024),
+		})
+		require.Nil(t, err)
+	}
+
+	// Abort the transaction.
+	_ = txn.Abort(context.Background())
+
+	// Expectation: The consumer should not receive any messages.
+	done := make(chan struct{})
+	go func() {
+		_, err := consumer.Receive(context.Background())
+		if err != nil {
+			close(done)
+		}
+	}()
+
+	select {
+	case <-done:
+		// The consumer should not receive any messages.
+		require.Fail(t, "The consumer should not receive any messages")
+	case <-time.After(time.Second):
+	}
+
+	// Clean up: Close the consumer and producer instances.
+	consumer.Close()
+	producer.Close()
 }