Repository: apache/pulsar-client-go
Pull request #984 (2023-03-07 13:49:48)

[feat][txn]Implement transactionImpl
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



Patch:
diff --git a/go.mod b/go.mod
index fe6f415..3e7299b 100644
--- a/go.mod
+++ b/go.mod
@@ -26,6 +26,7 @@ require (
 	go.uber.org/atomic v1.7.0
 	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
 )
 
diff --git a/go.sum b/go.sum
index 4e96e39..528c458 100644
--- a/go.sum
+++ b/go.sum
@@ -427,6 +427,7 @@ golang.org/x/text v0.3.7 h1:olpwvP2KacW1ZWvsR7uQhoyTYvKAupfQrRGBFM352Gk=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 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 h1:/5xXl8Y5W96D+TtHSlonuFqGHIWVuyCkGJLwGh9JJFs=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
diff --git a/perf/perf-consumer.go b/perf/perf-consumer.go
index 6b6e411..825de62 100644
--- a/perf/perf-consumer.go
+++ b/perf/perf-consumer.go
@@ -31,10 +31,11 @@ import (
 
 // ConsumeArgs define the parameters required by consume
 type ConsumeArgs struct {
-	Topic               string
-	SubscriptionName    string
-	ReceiverQueueSize   int
-	EnableBatchIndexAck bool
+	Topic                             string
+	SubscriptionName                  string
+	ReceiverQueueSize                 int
+	EnableBatchIndexAck               bool
+	EnableAutoScaledReceiverQueueSize bool
 }
 
 func newConsumerCommand() *cobra.Command {
@@ -57,6 +58,8 @@ func newConsumerCommand() *cobra.Command {
 	flags.StringVarP(&consumeArgs.SubscriptionName, "subscription", "s", "sub", "Subscription name")
 	flags.IntVarP(&consumeArgs.ReceiverQueueSize, "receiver-queue-size", "r", 1000, "Receiver queue size")
 	flags.BoolVar(&consumeArgs.EnableBatchIndexAck, "enable-batch-index-ack", false, "Whether to enable batch index ACK")
+	flags.BoolVar(&consumeArgs.EnableAutoScaledReceiverQueueSize, "enable-auto-scaled-queue-size", false,
+		"Whether to enable auto scaled receiver queue size")
 
 	return cmd
 }
@@ -76,9 +79,10 @@ func consume(consumeArgs *ConsumeArgs, stop <-chan struct{}) {
 	defer client.Close()
 
 	consumer, err := client.Subscribe(pulsar.ConsumerOptions{
-		Topic:                          consumeArgs.Topic,
-		SubscriptionName:               consumeArgs.SubscriptionName,
-		EnableBatchIndexAcknowledgment: consumeArgs.EnableBatchIndexAck,
+		Topic:                             consumeArgs.Topic,
+		SubscriptionName:                  consumeArgs.SubscriptionName,
+		EnableBatchIndexAcknowledgment:    consumeArgs.EnableBatchIndexAck,
+		EnableAutoScaledReceiverQueueSize: consumeArgs.EnableAutoScaledReceiverQueueSize,
 	})
 
 	if err != nil {
diff --git a/perf/perf-producer.go b/perf/perf-producer.go
index 9d062ae..f8e14d2 100644
--- a/perf/perf-producer.go
+++ b/perf/perf-producer.go
@@ -22,6 +22,8 @@ import (
 	"encoding/json"
 	"time"
 
+	"golang.org/x/time/rate"
+
 	"github.com/bmizerany/perks/quantile"
 	"github.com/spf13/cobra"
 
@@ -104,16 +106,9 @@ func produce(produceArgs *ProduceArgs, stop <-chan struct{}) {
 	payload := make([]byte, produceArgs.MessageSize)
 
 	ch := make(chan float64)
-	rateLimitCh := make(chan time.Time, produceArgs.Rate)
-	go func(rateLimit int, interval time.Duration) {
-		if rateLimit <= 0 { // 0 as no limit enforced
-			return
-		}
-		for {
-			oldest := <-rateLimitCh
-			time.Sleep(interval - time.Since(oldest))
-		}
-	}(produceArgs.Rate, time.Second)
+
+	limit := rate.Every(time.Duration(float64(time.Second) / float64(produceArgs.Rate)))
+	rateLimiter := rate.NewLimiter(limit, produceArgs.Rate)
 
 	go func(stopCh <-chan struct{}) {
 		for {
@@ -125,7 +120,7 @@ func produce(produceArgs *ProduceArgs, stop <-chan struct{}) {
 
 			start := time.Now()
 			if produceArgs.Rate > 0 {
-				rateLimitCh <- start
+				_ = rateLimiter.Wait(context.TODO())
 			}
 
 			producer.SendAsync(ctx, &pulsar.ProducerMessage{
diff --git a/pulsar/ack_grouping_tracker.go b/pulsar/ack_grouping_tracker.go
index dbc70f5..dd05186 100644
--- a/pulsar/ack_grouping_tracker.go
+++ b/pulsar/ack_grouping_tracker.go
@@ -18,8 +18,11 @@
 package pulsar
 
 import (
+	"sync"
+	"sync/atomic"
 	"time"
 
+	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
 	"github.com/bits-and-blooms/bitset"
 )
 
@@ -37,17 +40,10 @@ type ackGroupingTracker interface {
 	close()
 }
 
-type ackFlushType int
-
-const (
-	flushOnly ackFlushType = iota
-	flushAndClean
-	flushAndClose
-)
-
 func newAckGroupingTracker(options *AckGroupingOptions,
 	ackIndividual func(id MessageID),
-	ackCumulative func(id MessageID)) ackGroupingTracker {
+	ackCumulative func(id MessageID),
+	ackList func(ids []*pb.MessageIdData)) ackGroupingTracker {
 	if options == nil {
 		options = &AckGroupingOptions{
 			MaxSize: 1000,
@@ -62,66 +58,29 @@ func newAckGroupingTracker(options *AckGroupingOptions,
 		}
 	}
 
-	c := &cachedAcks{
-		singleAcks:        make([]MessageID, options.MaxSize),
-		pendingAcks:       make(map[int64]*bitset.BitSet),
-		lastCumulativeAck: EarliestMessageID(),
-		ackIndividual:     ackIndividual,
+	t := &timedAckGroupingTracker{
+		maxNumAcks:        int(options.MaxSize),
 		ackCumulative:     ackCumulative,
-		ackList: func(ids []MessageID) {
-			// TODO: support ack a list of MessageIDs
-			for _, id := range ids {
-				ackIndividual(id)
-			}
-		},
+		ackList:           ackList,
+		pendingAcks:       make(map[[2]uint64]*bitset.BitSet),
+		lastCumulativeAck: EarliestMessageID(),
 	}
 
-	timeout := time.NewTicker(time.Hour)
 	if options.MaxTime > 0 {
-		timeout = time.NewTicker(options.MaxTime)
-	} else {
-		timeout.Stop()
-	}
-	t := &timedAckGroupingTracker{
-		ackIndividualCh:   make(chan MessageID),
-		ackCumulativeCh:   make(chan MessageID),
-		duplicateIDCh:     make(chan MessageID),
-		duplicateResultCh: make(chan bool),
-		flushCh:           make(chan ackFlushType),
-		waitFlushCh:       make(chan bool),
-	}
-	go func() {
-		for {
-			select {
-			case id := <-t.ackIndividualCh:
-				if c.addAndCheckIfFull(id) {
-					c.flushIndividualAcks()
-					if options.MaxTime > 0 {
-						timeout.Reset(options.MaxTime)
-					}
-				}
-			case id := <-t.ackCumulativeCh:
-				c.tryUpdateLastCumulativeAck(id)
-				if options.MaxTime <= 0 {
-					c.flushCumulativeAck()
-				}
-			case id := <-t.duplicateIDCh:
-				t.duplicateResultCh <- c.isDuplicate(id)
-			case <-timeout.C:
-				c.flush()
-			case ackFlushType := <-t.flushCh:
-				timeout.Stop()
-				c.flush()
-				if ackFlushType == flushAndClean {
-					c.clean()
-				}
-				t.waitFlushCh <- true
-				if ackFlushType == flushAndClose {
+		t.ticker = time.NewTicker(options.MaxTime)
+		t.exitCh = make(chan struct{})
+		go func() {
+			for {
+				select {
+				case <-t.exitCh:
 					return
+				case <-t.ticker.C:
+					t.flush()
 				}
 			}
-		}
-	}()
+		}()
+	}
+
 	return t
 }
 
@@ -151,148 +110,157 @@ func (i *immediateAckGroupingTracker) flushAndClean() {
 func (i *immediateAckGroupingTracker) close() {
 }
 
-type cachedAcks struct {
-	singleAcks []MessageID
-	index      int
+type timedAckGroupingTracker struct {
+	sync.RWMutex
 
-	// Key is the hash code of the ledger id and the netry id,
+	maxNumAcks    int
+	ackCumulative func(id MessageID)
+	ackList       func(ids []*pb.MessageIdData)
+	ticker        *time.Ticker
+
+	// Key is the pair of the ledger id and the entry id,
 	// Value is the bit set that represents which messages are acknowledged if the entry stores a batch.
 	// The bit 1 represents the message has been acknowledged, i.e. the bits "111" represents all messages
 	// in the batch whose batch size is 3 are not acknowledged.
 	// After the 1st message (i.e. batch index is 0) is acknowledged, the bits will become "011".
 	// Value is nil if the entry represents a single message.
-	pendingAcks map[int64]*bitset.BitSet
+	pendingAcks map[[2]uint64]*bitset.BitSet
 
 	lastCumulativeAck     MessageID
-	cumulativeAckRequired bool
+	cumulativeAckRequired int32
 
-	ackIndividual func(id MessageID)
-	ackCumulative func(id MessageID)
-	ackList       func(ids []MessageID)
+	exitCh chan struct{}
 }
 
-func (t *cachedAcks) addAndCheckIfFull(id MessageID) bool {
-	t.singleAcks[t.index] = id
-	t.index++
-	key := messageIDHash(id)
-	ackSet, found := t.pendingAcks[key]
-	if !found {
-		if messageIDIsBatch(id) {
-			ackSet = bitset.New(uint(id.BatchSize()))
-			for i := 0; i < int(id.BatchSize()); i++ {
-				ackSet.Set(uint(i))
+func (t *timedAckGroupingTracker) add(id MessageID) {
+	if acks := t.tryAddIndividual(id); acks != nil {
+		t.flushIndividual(acks)
+	}
+}
+
+func (t *timedAckGroupingTracker) tryAddIndividual(id MessageID) map[[2]uint64]*bitset.BitSet {
+	t.Lock()
+	defer t.Unlock()
+	key := [2]uint64{uint64(id.LedgerID()), uint64(id.EntryID())}
+
+	batchIdx := id.BatchIdx()
+	batchSize := id.BatchSize()
+
+	if batchIdx >= 0 && batchSize > 0 {
+		bs, found := t.pendingAcks[key]
+		if !found {
+			if batchSize > 1 {
+				bs = bitset.New(uint(batchSize))
+				for i := uint(0); i < uint(batchSize); i++ {
+					bs.Set(i)
+				}
 			}
-			t.pendingAcks[key] = ackSet
-		} else {
-			t.pendingAcks[key] = nil
+			t.pendingAcks[key] = bs
+		}
+		if bs != nil {
+			bs.Clear(uint(batchIdx))
 		}
+	} else {
+		t.pendingAcks[key] = nil
 	}
-	if ackSet != nil {
-		ackSet.Clear(uint(id.BatchIdx()))
+
+	if len(t.pendingAcks) >= t.maxNumAcks {
+		pendingAcks := t.pendingAcks
+		t.pendingAcks = make(map[[2]uint64]*bitset.BitSet)
+		return pendingAcks
 	}
-	return t.index == len(t.singleAcks)
+	return nil
 }
 
-func (t *cachedAcks) tryUpdateLastCumulativeAck(id MessageID) {
+func (t *timedAckGroupingTracker) addCumulative(id MessageID) {
+	if t.tryUpdateCumulative(id) && t.ticker == nil {
+		t.ackCumulative(id)
+	}
+}
+
+func (t *timedAckGroupingTracker) tryUpdateCumulative(id MessageID) bool {
+	t.Lock()
+	defer t.Unlock()
 	if messageIDCompare(t.lastCumulativeAck, id) < 0 {
 		t.lastCumulativeAck = id
-		t.cumulativeAckRequired = true
+		atomic.StoreInt32(&t.cumulativeAckRequired, 1)
+		return true
 	}
+	return false
 }
 
-func (t *cachedAcks) isDuplicate(id MessageID) bool {
+func (t *timedAckGroupingTracker) isDuplicate(id MessageID) bool {
+	t.RLock()
+	defer t.RUnlock()
 	if messageIDCompare(t.lastCumulativeAck, id) >= 0 {
 		return true
 	}
-	ackSet, found := t.pendingAcks[messageIDHash(id)]
-	if !found {
-		return false
-	}
-	if ackSet == nil || !messageIDIsBatch(id) {
-		// NOTE: should we panic when ackSet != nil and messageIDIsBatch(id) is true?
-		return true
-	}
-	// 0 represents the message has been acknowledged
-	return !ackSet.Test(uint(id.BatchIdx()))
-}
-
-func (t *cachedAcks) flushIndividualAcks() {
-	if t.index > 0 {
-		t.ackList(t.singleAcks[0:t.index])
-		for _, id := range t.singleAcks[0:t.index] {
-			key := messageIDHash(id)
-			ackSet, found := t.pendingAcks[key]
-			if !found {
-				continue
-			}
-			if ackSet == nil {
-				delete(t.pendingAcks, key)
-			} else {
-				ackSet.Clear(uint(id.BatchIdx()))
-				if ackSet.None() { // all messages have been acknowledged
-					delete(t.pendingAcks, key)
-				}
-			}
-			delete(t.pendingAcks, messageIDHash(id))
+	key := [2]uint64{uint64(id.LedgerID()), uint64(id.EntryID())}
+	if bs, found := t.pendingAcks[key]; found {
+		if bs == nil {
+			return true
+		}
+		if !bs.Test(uint(id.BatchIdx())) {
+			return true
 		}
-		t.index = 0
 	}
+	return false
 }
 
-func (t *cachedAcks) flushCumulativeAck() {
-	if t.cumulativeAckRequired {
-		t.ackCumulative(t.lastCumulativeAck)
-		t.cumulativeAckRequired = false
+func (t *timedAckGroupingTracker) flush() {
+	if acks := t.clearPendingAcks(); len(acks) > 0 {
+		t.flushIndividual(acks)
+	}
+	if atomic.CompareAndSwapInt32(&t.cumulativeAckRequired, 1, 0) {
+		t.RLock()
+		id := t.lastCumulativeAck
+		t.RUnlock()
+		t.ackCumulative(id)
 	}
 }
 
-func (t *cachedAcks) flush() {
-	t.flushIndividualAcks()
-	t.flushCumulativeAck()
-}
-
-func (t *cachedAcks) clean() {
-	maxSize := len(t.singleAcks)
-	t.singleAcks = make([]MessageID, maxSize)
-	t.index = 0
-	t.pendingAcks = make(map[int64]*bitset.BitSet)
-	t.lastCumulativeAck = EarliestMessageID()
-	t.cumulativeAckRequired = false
-}
-
-type timedAckGroupingTracker struct {
-	ackIndividualCh   chan MessageID
-	ackCumulativeCh   chan MessageID
-	duplicateIDCh     chan MessageID
-	duplicateResultCh chan bool
-	flushCh           chan ackFlushType
-	waitFlushCh       chan bool
-}
-
-func (t *timedAckGroupingTracker) add(id MessageID) {
-	t.ackIndividualCh <- id
-}
-
-func (t *timedAckGroupingTracker) addCumulative(id MessageID) {
-	t.ackCumulativeCh <- id
-}
-
-func (t *timedAckGroupingTracker) isDuplicate(id MessageID) bool {
-	t.duplicateIDCh <- id
-	return <-t.duplicateResultCh
+func (t *timedAckGroupingTracker) flushAndClean() {
+	if acks := t.clearPendingAcks(); len(acks) > 0 {
+		t.flushIndividual(acks)
+	}
+	if atomic.CompareAndSwapInt32(&t.cumulativeAckRequired, 1, 0) {
+		t.Lock()
+		id := t.lastCumulativeAck
+		t.lastCumulativeAck = EarliestMessageID()
+		t.Unlock()
+		t.ackCumulative(id)
+	}
 }
 
-func (t *timedAckGroupingTracker) flush() {
-	t.flushCh <- flushOnly
-	<-t.waitFlushCh
+func (t *timedAckGroupingTracker) clearPendingAcks() map[[2]uint64]*bitset.BitSet {
+	t.Lock()
+	defer t.Unlock()
+	pendingAcks := t.pendingAcks
+	t.pendingAcks = make(map[[2]uint64]*bitset.BitSet)
+	return pendingAcks
 }
 
-func (t *timedAckGroupingTracker) flushAndClean() {
-	t.flushCh <- flushAndClean
-	<-t.waitFlushCh
+func (t *timedAckGroupingTracker) close() {
+	t.flushAndClean()
+	if t.exitCh != nil {
+		close(t.exitCh)
+	}
 }
 
-func (t *timedAckGroupingTracker) close() {
-	t.flushCh <- flushAndClose
-	<-t.waitFlushCh
+func (t *timedAckGroupingTracker) flushIndividual(pendingAcks map[[2]uint64]*bitset.BitSet) {
+	msgIDs := make([]*pb.MessageIdData, 0, len(pendingAcks))
+	for k, v := range pendingAcks {
+		ledgerID := k[0]
+		entryID := k[1]
+		msgID := &pb.MessageIdData{LedgerId: &ledgerID, EntryId: &entryID}
+		if v != nil && !v.None() {
+			bytes := v.Bytes()
+			msgID.AckSet = make([]int64, len(bytes))
+			for i := 0; i < len(bytes); i++ {
+				msgID.AckSet[i] = int64(bytes[i])
+			}
+		}
+		msgIDs = append(msgIDs, msgID)
+	}
+	t.ackList(msgIDs)
 }
diff --git a/pulsar/client_impl.go b/pulsar/client_impl.go
index 7c8fcc9..5322597 100644
--- a/pulsar/client_impl.go
+++ b/pulsar/client_impl.go
@@ -30,12 +30,13 @@ import (
 )
 
 const (
-	defaultConnectionTimeout = 10 * time.Second
-	defaultOperationTimeout  = 30 * time.Second
-	defaultKeepAliveInterval = 30 * time.Second
-	defaultMemoryLimitBytes  = 64 * 1024 * 1024
-	defaultConnMaxIdleTime   = 180 * time.Second
-	minConnMaxIdleTime       = 60 * time.Second
+	defaultConnectionTimeout           = 10 * time.Second
+	defaultOperationTimeout            = 30 * time.Second
+	defaultKeepAliveInterval           = 30 * time.Second
+	defaultMemoryLimitBytes            = 64 * 1024 * 1024
+	defaultMemoryLimitTriggerThreshold = 0.95
+	defaultConnMaxIdleTime             = 180 * time.Second
+	minConnMaxIdleTime                 = 60 * time.Second
 )
 
 type client struct {
@@ -158,7 +159,7 @@ func newClient(options ClientOptions) (Client, error) {
 			maxConnectionsPerHost, logger, metrics, connectionMaxIdleTime),
 		log:      logger,
 		metrics:  metrics,
-		memLimit: internal.NewMemoryLimitController(memLimitBytes),
+		memLimit: internal.NewMemoryLimitController(memLimitBytes, defaultMemoryLimitTriggerThreshold),
 	}
 	serviceNameResolver := internal.NewPulsarServiceNameResolver(url)
 
diff --git a/pulsar/consumer.go b/pulsar/consumer.go
index 9576d7a..64a096d 100644
--- a/pulsar/consumer.go
+++ b/pulsar/consumer.go
@@ -160,6 +160,12 @@ type ConsumerOptions struct {
 	// Default value is `1000` messages and should be good for most use cases.
 	ReceiverQueueSize int
 
+	// EnableAutoScaledReceiverQueueSize, if enabled, the consumer receive queue will be auto-scaled
+	// by the consumer actual throughput. The ReceiverQueueSize will be the maximum size which consumer
+	// receive queue can be scaled.
+	// Default is false.
+	EnableAutoScaledReceiverQueueSize bool
+
 	// NackRedeliveryDelay specifies the delay after which to redeliver the messages that failed to be
 	// processed. Default is 1 min. (See `Consumer.Nack()`)
 	NackRedeliveryDelay time.Duration
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
index d19e522..fd7fa57 100644
--- a/pulsar/consumer_impl.go
+++ b/pulsar/consumer_impl.go
@@ -399,6 +399,7 @@ func (c *consumer) internalTopicSubscribeToPartitions() error {
 				consumerEventListener:       c.options.EventListener,
 				enableBatchIndexAck:         c.options.EnableBatchIndexAcknowledgment,
 				ackGroupingOptions:          c.options.AckGroupingOptions,
+				autoReceiverQueueSize:       c.options.EnableAutoScaledReceiverQueueSize,
 			}
 			cons, err := newPartitionConsumer(c, c.client, opts, c.messageCh, c.dlq, c.metrics)
 			ch <- ConsumerError{
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index a3dac19..fb77d0d 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -25,7 +25,6 @@ import (
 	"math"
 	"strings"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"google.golang.org/protobuf/proto"
@@ -77,6 +76,11 @@ const (
 	nonDurable
 )
 
+const (
+	initialReceiverQueueSize           = 1
+	receiverQueueExpansionMemThreshold = 0.75
+)
+
 const (
 	noMessageEntry = -1
 )
@@ -89,6 +93,7 @@ type partitionConsumerOpts struct {
 	subscriptionInitPos         SubscriptionInitialPosition
 	partitionIdx                int
 	receiverQueueSize           int
+	autoReceiverQueueSize       bool
 	nackRedeliveryDelay         time.Duration
 	nackBackoffPolicy           NackBackoffPolicy
 	metadata                    map[string]string
@@ -142,11 +147,15 @@ type partitionConsumer struct {
 	availablePermits *availablePermits
 
 	// the size of the queue channel for buffering messages
-	queueSize       int32
+	maxQueueSize    int32
 	queueCh         chan []*message
 	startMessageID  atomicMessageID
 	lastDequeuedMsg *trackingMessageID
 
+	currentQueueSize       uAtomic.Int32
+	scaleReceiverQueueHint uAtomic.Bool
+	incomingMessages       uAtomic.Int32
+
 	eventsCh        chan interface{}
 	connectedCh     chan struct{}
 	connectClosedCh chan connectionClosed
@@ -181,22 +190,44 @@ func (pc *partitionConsumer) ActiveConsumerChanged(isActive bool) {
 }
 
 type availablePermits struct {
-	permits int32
+	permits uAtomic.Int32
 	pc      *partitionConsumer
 }
 
 func (p *availablePermits) inc() {
 	// atomic add availablePermits
-	ap := atomic.AddInt32(&p.permits, 1)
+	p.add(1)
+}
+
+func (p *availablePermits) add(delta int32) {
+	p.permits.Add(delta)
+	p.flowIfNeed()
+}
 
+func (p *availablePermits) reset() {
+	p.permits.Store(0)
+}
+
+func (p *availablePermits) get() int32 {
+	return p.permits.Load()
+}
+
+func (p *availablePermits) flowIfNeed() {
 	// TODO implement a better flow controller
 	// send more permits if needed
-	flowThreshold := int32(math.Max(float64(p.pc.queueSize/2), 1))
-	if ap >= flowThreshold {
-		availablePermits := ap
-		requestedPermits := ap
+	var flowThreshold int32
+	if p.pc.options.autoReceiverQueueSize {
+		flowThreshold = int32(math.Max(float64(p.pc.currentQueueSize.Load()/2), 1))
+	} else {
+		flowThreshold = int32(math.Max(float64(p.pc.maxQueueSize/2), 1))
+	}
+
+	current := p.get()
+	if current >= flowThreshold {
+		availablePermits := current
+		requestedPermits := current
 		// check if permits changed
-		if !atomic.CompareAndSwapInt32(&p.permits, ap, 0) {
+		if !p.permits.CAS(current, 0) {
 			return
 		}
 
@@ -207,10 +238,6 @@ func (p *availablePermits) inc() {
 	}
 }
 
-func (p *availablePermits) reset() {
-	atomic.StoreInt32(&p.permits, 0)
-}
-
 // atomicMessageID is a wrapper for trackingMessageID to make get and set atomic
 type atomicMessageID struct {
 	msgID *trackingMessageID
@@ -292,7 +319,7 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 		consumerID:           client.rpcClient.NewConsumerID(),
 		partitionIdx:         int32(options.partitionIdx),
 		eventsCh:             make(chan interface{}, 10),
-		queueSize:            int32(options.receiverQueueSize),
+		maxQueueSize:         int32(options.receiverQueueSize),
 		queueCh:              make(chan []*message, options.receiverQueueSize),
 		startMessageID:       atomicMessageID{msgID: options.startMessageID},
 		connectedCh:          make(chan struct{}),
@@ -305,12 +332,19 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 		metrics:              metrics,
 		schemaInfoCache:      newSchemaInfoCache(client, options.topic),
 	}
+	if pc.options.autoReceiverQueueSize {
+		pc.currentQueueSize.Store(initialReceiverQueueSize)
+		pc.client.memLimit.RegisterTrigger(pc.shrinkReceiverQueueSize)
+	} else {
+		pc.currentQueueSize.Store(int32(pc.options.receiverQueueSize))
+	}
 	pc.availablePermits = &availablePermits{pc: pc}
 	pc.chunkedMsgCtxMap = newChunkedMsgCtxMap(options.maxPendingChunkedMessage, pc)
 	pc.unAckChunksTracker = newUnAckChunksTracker(pc)
 	pc.ackGroupingTracker = newAckGroupingTracker(options.ackGroupingOptions,
 		func(id MessageID) { pc.sendIndividualAck(id) },
-		func(id MessageID) { pc.sendCumulativeAck(id) })
+		func(id MessageID) { pc.sendCumulativeAck(id) },
+		func(ids []*pb.MessageIdData) { pc.eventsCh <- ids })
 	pc.setConsumerState(consumerInit)
 	pc.log = client.log.SubLogger(log.Fields{
 		"name":         pc.name,
@@ -467,6 +501,14 @@ func (pc *partitionConsumer) ackID(msgID MessageID, withResponse bool) error {
 	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 {
@@ -837,6 +879,14 @@ func (pc *partitionConsumer) internalAck(req *ackRequest) {
 	}
 }
 
+func (pc *partitionConsumer) internalAckList(msgIDs []*pb.MessageIdData) {
+	pc.client.rpcClient.RequestOnCnxNoWait(pc._getConn(), pb.BaseCommand_ACK, &pb.CommandAck{
+		AckType:    pb.CommandAck_Individual.Enum(),
+		ConsumerId: proto.Uint64(pc.consumerID),
+		MessageId:  msgIDs,
+	})
+}
+
 func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, headersAndPayload internal.Buffer) error {
 	pbMsgID := response.GetMessageId()
 
@@ -895,6 +945,12 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 					orderingKey:         string(msgMeta.OrderingKey),
 				},
 			}
+
+			if pc.options.autoReceiverQueueSize {
+				pc.incomingMessages.Inc()
+				pc.markScaleIfNeed()
+			}
+
 			pc.queueCh <- messages
 			return nil
 		}
@@ -948,6 +1004,7 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 	pc.metrics.MessagesReceived.Add(float64(numMsgs))
 	pc.metrics.PrefetchedMessages.Add(float64(numMsgs))
 
+	var bytesReceived int
 	for i := 0; i < numMsgs; i++ {
 		smm, payload, err := reader.ReadMessage()
 		if err != nil || payload == nil {
@@ -1062,6 +1119,13 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 		})
 
 		messages = append(messages, msg)
+		bytesReceived += msg.size()
+	}
+
+	if pc.options.autoReceiverQueueSize {
+		pc.client.memLimit.ForceReserveMemory(int64(bytesReceived))
+		pc.incomingMessages.Add(int32(len(messages)))
+		pc.markScaleIfNeed()
 	}
 
 	// send messages to the dispatcher
@@ -1211,6 +1275,7 @@ func (pc *partitionConsumer) dispatcher() {
 		var queueCh chan []*message
 		var messageCh chan ConsumerMessage
 		var nextMessage ConsumerMessage
+		var nextMessageSize int
 
 		// are there more messages to send?
 		if len(messages) > 0 {
@@ -1218,6 +1283,7 @@ func (pc *partitionConsumer) dispatcher() {
 				Consumer: pc.parentConsumer,
 				Message:  messages[0],
 			}
+			nextMessageSize = messages[0].size()
 
 			if pc.dlq.shouldSendToDlq(&nextMessage) {
 				// pass the message to the DLQ router
@@ -1231,7 +1297,6 @@ func (pc *partitionConsumer) dispatcher() {
 			pc.metrics.PrefetchedMessages.Dec()
 			pc.metrics.PrefetchedBytes.Sub(float64(len(messages[0].payLoad)))
 		} else {
-			// we are ready for more messages
 			queueCh = pc.queueCh
 		}
 
@@ -1249,7 +1314,13 @@ func (pc *partitionConsumer) dispatcher() {
 
 			// reset available permits
 			pc.availablePermits.reset()
-			initialPermits := uint32(pc.queueSize)
+
+			var initialPermits uint32
+			if pc.options.autoReceiverQueueSize {
+				initialPermits = uint32(pc.currentQueueSize.Load())
+			} else {
+				initialPermits = uint32(pc.maxQueueSize)
+			}
 
 			pc.log.Debugf("dispatcher requesting initial permits=%d", initialPermits)
 			// send initial permits
@@ -1273,6 +1344,12 @@ func (pc *partitionConsumer) dispatcher() {
 
 			pc.availablePermits.inc()
 
+			if pc.options.autoReceiverQueueSize {
+				pc.incomingMessages.Dec()
+				pc.client.memLimit.ReleaseMemory(int64(nextMessageSize))
+				pc.expectMoreIncomingMessages()
+			}
+
 		case clearQueueCb := <-pc.clearQueueCh:
 			// drain the message queue on any new connection by sending a
 			// special nil message to the channel so we know when to stop dropping messages
@@ -1288,6 +1365,9 @@ func (pc *partitionConsumer) dispatcher() {
 				} else if nextMessageInQueue == nil {
 					nextMessageInQueue = toTrackingMessageID(m[0].msgID)
 				}
+				if pc.options.autoReceiverQueueSize {
+					pc.incomingMessages.Sub(int32(len(m)))
+				}
 			}
 
 			messages = nil
@@ -1364,6 +1444,8 @@ func (pc *partitionConsumer) runEventsLoop() {
 			switch v := i.(type) {
 			case *ackRequest:
 				pc.internalAck(v)
+			case []*pb.MessageIdData:
+				pc.internalAckList(v)
 			case *redeliveryRequest:
 				pc.internalRedeliver(v)
 			case *unsubscribeRequest:
@@ -1660,6 +1742,48 @@ func getPreviousMessage(mid *trackingMessageID) *trackingMessageID {
 	}
 }
 
+func (pc *partitionConsumer) expectMoreIncomingMessages() {
+	if !pc.options.autoReceiverQueueSize {
+		return
+	}
+	if pc.scaleReceiverQueueHint.CAS(true, false) {
+		oldSize := pc.currentQueueSize.Load()
+		maxSize := int32(pc.options.receiverQueueSize)
+		newSize := int32(math.Min(float64(maxSize), float64(oldSize*2)))
+		usagePercent := pc.client.memLimit.CurrentUsagePercent()
+		if usagePercent < receiverQueueExpansionMemThreshold && newSize > oldSize {
+			pc.currentQueueSize.CAS(oldSize, newSize)
+			pc.availablePermits.add(newSize - oldSize)
+			pc.log.Debugf("update currentQueueSize from %d -> %d", oldSize, newSize)
+		}
+	}
+}
+
+func (pc *partitionConsumer) markScaleIfNeed() {
+	// availablePermits + incomingMessages (messages in queueCh) is the number of prefetched messages
+	// The result of auto-scale we expected is currentQueueSize is slightly bigger than prefetched messages
+	prev := pc.scaleReceiverQueueHint.Swap(pc.availablePermits.get()+pc.incomingMessages.Load() >=
+		pc.currentQueueSize.Load())
+	if prev != pc.scaleReceiverQueueHint.Load() {
+		pc.log.Debugf("update scaleReceiverQueueHint from %t -> %t", prev, pc.scaleReceiverQueueHint.Load())
+	}
+}
+
+func (pc *partitionConsumer) shrinkReceiverQueueSize() {
+	if !pc.options.autoReceiverQueueSize {
+		return
+	}
+
+	oldSize := pc.currentQueueSize.Load()
+	minSize := int32(math.Min(float64(initialReceiverQueueSize), float64(pc.options.receiverQueueSize)))
+	newSize := int32(math.Max(float64(minSize), float64(oldSize/2)))
+	if newSize < oldSize {
+		pc.currentQueueSize.CAS(oldSize, newSize)
+		pc.availablePermits.add(newSize - oldSize)
+		pc.log.Debugf("update currentQueueSize from %d -> %d", oldSize, newSize)
+	}
+}
+
 func (pc *partitionConsumer) Decompress(msgMeta *pb.MessageMetadata, payload internal.Buffer) (internal.Buffer, error) {
 	providerEntry, ok := pc.compressionProviders.Load(msgMeta.GetCompression())
 	if !ok {
diff --git a/pulsar/dlq_router.go b/pulsar/dlq_router.go
index 5ecd8f8..6d30003 100644
--- a/pulsar/dlq_router.go
+++ b/pulsar/dlq_router.go
@@ -109,13 +109,17 @@ func (r *dlqRouter) run() {
 				Properties:          properties,
 				EventTime:           msg.EventTime(),
 				ReplicationClusters: msg.replicationClusters,
-			}, func(MessageID, *ProducerMessage, error) {
-				r.log.WithField("msgID", msgID).Debug("Sent message to DLQ")
-
-				// The Producer ack might be coming from the connection go-routine that
-				// is also used by the consumer. In that case we would get a dead-lock
-				// if we'd try to ack.
-				go cm.Consumer.AckID(msgID)
+			}, func(messageID MessageID, producerMessage *ProducerMessage, err error) {
+				if err == nil {
+					r.log.WithField("msgID", msgID).Debug("Succeed to send message to DLQ")
+					// The Producer ack might be coming from the connection go-routine that
+					// is also used by the consumer. In that case we would get a dead-lock
+					// if we'd try to ack.
+					go cm.Consumer.AckID(msgID)
+				} else {
+					r.log.WithError(err).WithField("msgID", msgID).Debug("Failed to send message to DLQ")
+					go cm.Consumer.Nack(cm)
+				}
 			})
 
 		case <-r.closeCh:
diff --git a/pulsar/error.go b/pulsar/error.go
index 0aa1e3c..73a0b60 100644
--- a/pulsar/error.go
+++ b/pulsar/error.go
@@ -17,7 +17,11 @@
 
 package pulsar
 
-import "fmt"
+import (
+	"fmt"
+
+	proto "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
+)
 
 // Result used to represent pulsar processing is an alias of type int.
 type Result int
@@ -103,14 +107,11 @@ const (
 	ProducerClosed
 	// SchemaFailure means the payload could not be encoded using the Schema
 	SchemaFailure
-
-	// ReachMaxPendingOps means the pending operations in transaction_impl coordinator reach the maximum.
-	ReachMaxPendingOps
 	// InvalidStatus means the component status is not as expected.
 	InvalidStatus
-	// TransactionError means this is a transaction related error
-	TransactionError
-
+	// TransactionNoFoundError The transaction is not exist in the transaction coordinator, It may be an error txn
+	// or already ended.
+	TransactionNoFoundError
 	// ClientMemoryBufferIsFull client limit buffer is full
 	ClientMemoryBufferIsFull
 )
@@ -221,7 +222,20 @@ func getResultStr(r Result) string {
 		return "SchemaFailure"
 	case ClientMemoryBufferIsFull:
 		return "ClientMemoryBufferIsFull"
+	case TransactionNoFoundError:
+		return "TransactionNoFoundError"
 	default:
 		return fmt.Sprintf("Result(%d)", r)
 	}
 }
+
+func getErrorFromServerError(serverError *proto.ServerError) error {
+	switch *serverError {
+	case proto.ServerError_TransactionNotFound:
+		return newError(TransactionNoFoundError, serverError.String())
+	case proto.ServerError_InvalidTxnStatus:
+		return newError(InvalidStatus, serverError.String())
+	default:
+		return newError(UnknownError, serverError.String())
+	}
+}
diff --git a/pulsar/impl_message.go b/pulsar/impl_message.go
index 9c56070..89a709f 100644
--- a/pulsar/impl_message.go
+++ b/pulsar/impl_message.go
@@ -382,6 +382,10 @@ func (msg *message) BrokerPublishTime() *time.Time {
 	return msg.brokerPublishTime
 }
 
+func (msg *message) size() int {
+	return len(msg.payLoad)
+}
+
 func newAckTracker(size uint) *ackTracker {
 	batchIDs := bitset.New(size)
 	for i := uint(0); i < size; i++ {
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index 55d04f3..59aad16 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -525,7 +525,7 @@ func (c *connection) internalReceivedCommand(cmd *pb.BaseCommand, headersAndPayl
 		c.handleResponse(cmd.Success.GetRequestId(), cmd)
 
 	case pb.BaseCommand_PRODUCER_SUCCESS:
-		if !*cmd.ProducerSuccess.ProducerReady {
+		if !cmd.ProducerSuccess.GetProducerReady() {
 			request, ok := c.findPendingRequest(cmd.ProducerSuccess.GetRequestId())
 			if ok {
 				request.callback(cmd, nil)
diff --git a/pulsar/internal/memory_limit_controller.go b/pulsar/internal/memory_limit_controller.go
index 5bf8d59..5ead9f5 100644
--- a/pulsar/internal/memory_limit_controller.go
+++ b/pulsar/internal/memory_limit_controller.go
@@ -31,18 +31,42 @@ type MemoryLimitController interface {
 	CurrentUsage() int64
 	CurrentUsagePercent() float64
 	IsMemoryLimited() bool
+	RegisterTrigger(trigger func())
 }
 
 type memoryLimitController struct {
 	limit        int64
 	chCond       *chCond
 	currentUsage int64
+
+	triggers []*thresholdTrigger
+	// valid range is (0, 1.0)
+	triggerThreshold float64
+}
+
+type thresholdTrigger struct {
+	triggerFunc    func()
+	triggerRunning int32
+}
+
+func (t *thresholdTrigger) canTryRunning() bool {
+	return atomic.CompareAndSwapInt32(&t.triggerRunning, 0, 1)
+}
+
+func (t *thresholdTrigger) setRunning(isRunning bool) {
+	if isRunning {
+		atomic.StoreInt32(&t.triggerRunning, 1)
+	} else {
+		atomic.StoreInt32(&t.triggerRunning, 0)
+	}
 }
 
-func NewMemoryLimitController(limit int64) MemoryLimitController {
+// NewMemoryLimitController threshold valid range is (0, 1.0)
+func NewMemoryLimitController(limit int64, threshold float64) MemoryLimitController {
 	mlc := &memoryLimitController{
-		limit:  limit,
-		chCond: newCond(&sync.Mutex{}),
+		limit:            limit,
+		chCond:           newCond(&sync.Mutex{}),
+		triggerThreshold: threshold,
 	}
 	return mlc
 }
@@ -72,13 +96,16 @@ func (m *memoryLimitController) TryReserveMemory(size int64) bool {
 		}
 
 		if atomic.CompareAndSwapInt64(&m.currentUsage, current, newUsage) {
+			m.checkTrigger(current, newUsage)
 			return true
 		}
 	}
 }
 
 func (m *memoryLimitController) ForceReserveMemory(size int64) {
-	atomic.AddInt64(&m.currentUsage, size)
+	nextUsage := atomic.AddInt64(&m.currentUsage, size)
+	prevUsage := nextUsage - size
+	m.checkTrigger(prevUsage, nextUsage)
 }
 
 func (m *memoryLimitController) ReleaseMemory(size int64) {
@@ -99,3 +126,26 @@ func (m *memoryLimitController) CurrentUsagePercent() float64 {
 func (m *memoryLimitController) IsMemoryLimited() bool {
 	return m.limit > 0
 }
+
+func (m *memoryLimitController) RegisterTrigger(trigger func()) {
+	m.chCond.L.Lock()
+	defer m.chCond.L.Unlock()
+	m.triggers = append(m.triggers, &thresholdTrigger{
+		triggerFunc: trigger,
+	})
+}
+
+func (m *memoryLimitController) checkTrigger(prevUsage int64, nextUsage int64) {
+	nextUsagePercent := float64(nextUsage) / float64(m.limit)
+	prevUsagePercent := float64(prevUsage) / float64(m.limit)
+	if nextUsagePercent >= m.triggerThreshold && prevUsagePercent < m.triggerThreshold {
+		for _, trigger := range m.triggers {
+			if trigger.canTryRunning() {
+				go func(trigger *thresholdTrigger) {
+					trigger.triggerFunc()
+					trigger.setRunning(false)
+				}(trigger)
+			}
+		}
+	}
+}
diff --git a/pulsar/internal/rpc_client.go b/pulsar/internal/rpc_client.go
index 0ee8ca9..2213083 100644
--- a/pulsar/internal/rpc_client.go
+++ b/pulsar/internal/rpc_client.go
@@ -138,7 +138,7 @@ func (c *rpcClient) Request(logicalAddr *url.URL, physicalAddr *url.URL, request
 			// Ignoring producer not ready response.
 			// Continue to wait for the producer to create successfully
 			if res.error == nil && *res.RPCResult.Response.Type == pb.BaseCommand_PRODUCER_SUCCESS {
-				if !*res.RPCResult.Response.ProducerSuccess.ProducerReady {
+				if !res.RPCResult.Response.ProducerSuccess.GetProducerReady() {
 					timeoutCh = nil
 					break
 				}
diff --git a/pulsar/message.go b/pulsar/message.go
index c44957d..98190e9 100644
--- a/pulsar/message.go
+++ b/pulsar/message.go
@@ -212,11 +212,3 @@ func messageIDCompare(lhs MessageID, rhs MessageID) int {
 	}
 	return 0
 }
-
-func messageIDHash(id MessageID) int64 {
-	return id.LedgerID() + 31*id.EntryID()
-}
-
-func messageIDIsBatch(id MessageID) bool {
-	return id.BatchIdx() >= 0 && id.BatchSize() > 0
-}
diff --git a/pulsar/reader.go b/pulsar/reader.go
index e4679ab..d58d06f 100644
--- a/pulsar/reader.go
+++ b/pulsar/reader.go
@@ -92,6 +92,16 @@ type ReaderOptions struct {
 	// BackoffPolicy parameterize the following options in the reconnection logic to
 	// allow users to customize the reconnection logic (minBackoff, maxBackoff and jitterPercentage)
 	BackoffPolicy internal.BackoffPolicy
+
+	// MaxPendingChunkedMessage sets the maximum pending chunked messages. (default: 100)
+	MaxPendingChunkedMessage int
+
+	// ExpireTimeOfIncompleteChunk sets the expiry time of discarding incomplete chunked message. (default: 60 seconds)
+	ExpireTimeOfIncompleteChunk time.Duration
+
+	// AutoAckIncompleteChunk sets whether reader auto acknowledges incomplete chunked message when it should
+	// be removed (e.g.the chunked message pending queue is full). (default: false)
+	AutoAckIncompleteChunk bool
 }
 
 // Reader can be used to scan through all the messages currently available in a topic.
diff --git a/pulsar/reader_impl.go b/pulsar/reader_impl.go
index c7620ad..36b492a 100644
--- a/pulsar/reader_impl.go
+++ b/pulsar/reader_impl.go
@@ -90,22 +90,33 @@ func newReader(client *client, options ReaderOptions) (Reader, error) {
 		options.Decryption.MessageCrypto = messageCrypto
 	}
 
+	if options.MaxPendingChunkedMessage == 0 {
+		options.MaxPendingChunkedMessage = 100
+	}
+
+	if options.ExpireTimeOfIncompleteChunk == 0 {
+		options.ExpireTimeOfIncompleteChunk = time.Minute
+	}
+
 	consumerOptions := &partitionConsumerOpts{
-		topic:                      options.Topic,
-		consumerName:               options.Name,
-		subscription:               subscriptionName,
-		subscriptionType:           Exclusive,
-		receiverQueueSize:          receiverQueueSize,
-		startMessageID:             startMessageID,
-		startMessageIDInclusive:    options.StartMessageIDInclusive,
-		subscriptionMode:           nonDurable,
-		readCompacted:              options.ReadCompacted,
-		metadata:                   options.Properties,
-		nackRedeliveryDelay:        defaultNackRedeliveryDelay,
-		replicateSubscriptionState: false,
-		decryption:                 options.Decryption,
-		schema:                     options.Schema,
-		backoffPolicy:              options.BackoffPolicy,
+		topic:                       options.Topic,
+		consumerName:                options.Name,
+		subscription:                subscriptionName,
+		subscriptionType:            Exclusive,
+		receiverQueueSize:           receiverQueueSize,
+		startMessageID:              startMessageID,
+		startMessageIDInclusive:     options.StartMessageIDInclusive,
+		subscriptionMode:            nonDurable,
+		readCompacted:               options.ReadCompacted,
+		metadata:                    options.Properties,
+		nackRedeliveryDelay:         defaultNackRedeliveryDelay,
+		replicateSubscriptionState:  false,
+		decryption:                  options.Decryption,
+		schema:                      options.Schema,
+		backoffPolicy:               options.BackoffPolicy,
+		maxPendingChunkedMessage:    options.MaxPendingChunkedMessage,
+		expireTimeOfIncompleteChunk: options.ExpireTimeOfIncompleteChunk,
+		autoAckIncompleteChunk:      options.AutoAckIncompleteChunk,
 	}
 
 	reader := &reader{
diff --git a/pulsar/transaction.go b/pulsar/transaction.go
index ae7c673..60e1d2b 100644
--- a/pulsar/transaction.go
+++ b/pulsar/transaction.go
@@ -17,7 +17,53 @@
 
 package pulsar
 
+import (
+	"context"
+)
+
+// TxnState The state of the transaction. Check the state of the transaction before executing some operation
+// with the transaction is necessary.
+type TxnState int32
+
+const (
+	_ TxnState = iota
+	// TxnOpen The transaction in TxnOpen state can be used to send/ack messages.
+	TxnOpen
+	// TxnCommitting The state of the transaction will be TxnCommitting after the commit method is called.
+	// The transaction in TxnCommitting state can be committed again.
+	TxnCommitting
+	// TxnAborting The state of the transaction will be TxnAborting after the abort method is called.
+	// The transaction in TxnAborting state can be aborted again.
+	TxnAborting
+	// TxnCommitted The state of the transaction will be TxnCommitted after the commit method is executed success.
+	// This means that all the operations with the transaction are success.
+	TxnCommitted
+	// TxnAborted The state of the transaction will be TxnAborted after the abort method is executed success.
+	// This means that all the operations with the transaction are aborted.
+	TxnAborted
+	// TxnError The state of the transaction will be TxnError after the operation of transaction get a non-retryable error.
+	TxnError
+	// TxnTimeout The state of the transaction will be TxnTimeout after the transaction timeout.
+	TxnTimeout
+)
+
+// TxnID An identifier for representing a transaction.
 type TxnID struct {
-	mostSigBits  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
+type Transaction interface {
+	//Commit You can commit the transaction after all the sending/acknowledging operations with the transaction success.
+	Commit(context.Context) error
+	//Abort You can abort the transaction when you want to abort all the sending/acknowledging operations
+	// with the transaction.
+	Abort(context.Context) error
+	//GetState Get the state of the transaction.
+	GetState() TxnState
+	//GetTxnID Get the identified ID of the transaction.
+	GetTxnID() TxnID
+}
diff --git a/pulsar/transaction_coordinator_client.go b/pulsar/transaction_coordinator_client.go
index 82d1490..1535fad 100644
--- a/pulsar/transaction_coordinator_client.go
+++ b/pulsar/transaction_coordinator_client.go
@@ -114,14 +114,16 @@ func (tc *transactionCoordinatorClient) newTransaction(timeout time.Duration) (*
 		TxnTtlSeconds: proto.Uint64(uint64(timeout.Milliseconds())),
 	}
 
-	cnx, err := tc.client.rpcClient.RequestOnCnx(tc.cons[nextTcID], requestID, pb.BaseCommand_NEW_TXN, cmdNewTxn)
+	res, err := tc.client.rpcClient.RequestOnCnx(tc.cons[nextTcID], requestID, pb.BaseCommand_NEW_TXN, cmdNewTxn)
 	tc.semaphore.Release()
 	if err != nil {
 		return nil, err
+	} else if res.Response.NewTxnResponse.Error != nil {
+		return nil, getErrorFromServerError(res.Response.NewTxnResponse.Error)
 	}
 
-	return &TxnID{*cnx.Response.NewTxnResponse.TxnidMostBits,
-		*cnx.Response.NewTxnResponse.TxnidLeastBits}, nil
+	return &TxnID{*res.Response.NewTxnResponse.TxnidMostBits,
+		*res.Response.NewTxnResponse.TxnidLeastBits}, nil
 }
 
 // addPublishPartitionToTxn register the partitions which published messages with the transactionImpl.
@@ -137,10 +139,15 @@ func (tc *transactionCoordinatorClient) addPublishPartitionToTxn(id *TxnID, part
 		TxnidLeastBits: proto.Uint64(id.leastSigBits),
 		Partitions:     partitions,
 	}
-	_, 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()
-	return err
+	if err != nil {
+		return err
+	} else if res.Response.AddPartitionToTxnResponse.Error != nil {
+		return getErrorFromServerError(res.Response.AddPartitionToTxnResponse.Error)
+	}
+	return nil
 }
 
 // addSubscriptionToTxn register the subscription which acked messages with the transactionImpl.
@@ -160,10 +167,15 @@ func (tc *transactionCoordinatorClient) addSubscriptionToTxn(id *TxnID, topic st
 		TxnidLeastBits: proto.Uint64(id.leastSigBits),
 		Subscription:   []*pb.Subscription{sub},
 	}
-	_, 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()
-	return err
+	if err != nil {
+		return err
+	} else if res.Response.AddSubscriptionToTxnResponse.Error != nil {
+		return getErrorFromServerError(res.Response.AddSubscriptionToTxnResponse.Error)
+	}
+	return nil
 }
 
 // endTxn commit or abort the transactionImpl.
@@ -178,9 +190,14 @@ func (tc *transactionCoordinatorClient) endTxn(id *TxnID, action pb.TxnAction) e
 		TxnidMostBits:  proto.Uint64(id.mostSigBits),
 		TxnidLeastBits: proto.Uint64(id.leastSigBits),
 	}
-	_, 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()
-	return err
+	if err != nil {
+		return err
+	} else if res.Response.EndTxnResponse.Error != nil {
+		return getErrorFromServerError(res.Response.EndTxnResponse.Error)
+	}
+	return nil
 }
 
 func getTCAssignTopicName(partition uint64) string {
diff --git a/pulsar/transaction_impl.go b/pulsar/transaction_impl.go
new file mode 100644
index 0000000..7cc93ec
--- /dev/null
+++ b/pulsar/transaction_impl.go
@@ -0,0 +1,237 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package pulsar
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
+	"github.com/apache/pulsar-client-go/pulsar/log"
+)
+
+type subscription struct {
+	topic        string
+	subscription string
+}
+
+type transaction struct {
+	sync.Mutex
+	txnID                    TxnID
+	state                    TxnState
+	tcClient                 *transactionCoordinatorClient
+	registerPartitions       map[string]bool
+	registerAckSubscriptions map[subscription]bool
+	// opsFlow It has two effects:
+	// 1. Wait all the operations of sending and acking messages with the transaction complete
+	// by reading msg from the chan.
+	// 2. Prevent sending or acking messages with a committed or aborted transaction.
+	// opsCount is record the number of the uncompleted operations.
+	// opsFlow
+	//   Write:
+	//     1. When the transaction is created, a bool will be written to opsFlow chan.
+	//     2. When the opsCount decrement from 1 to 0, a new bool will be written to opsFlow chan.
+	//     3. When get a retryable error after committing or aborting the transaction,
+	//        a bool will be written to opsFlow chan.
+	//   Read:
+	//     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
+	opTimeout time.Duration
+	log       log.Logger
+}
+
+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,
+		tcClient:                 tcClient,
+	}
+	//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
+		<-time.After(timeout)
+		atomic.CompareAndSwapInt32((*int32)(&transaction.state), int32(TxnOpen), int32(TxnTimeout))
+	}()
+	transaction.log = tcClient.log.SubLogger(log.Fields{})
+	return transaction
+}
+
+func (txn *transaction) GetState() TxnState {
+	return txn.state
+}
+
+func (txn *transaction) Commit(ctx 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())
+	}
+
+	//Wait for all operations to complete
+	select {
+	case <-txn.opsFlow:
+	case <-time.After(txn.opTimeout):
+		return newError(TimeoutError, "There are some operations that are not completed after the timeout.")
+	}
+	//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))
+	} else {
+		if err.(*Error).Result() == TransactionNoFoundError || err.(*Error).Result() == InvalidStatus {
+			atomic.StoreInt32((*int32)(&txn.state), int32(TxnError))
+			return err
+		}
+		txn.opsFlow <- true
+	}
+	return err
+}
+
+func (txn *transaction) Abort(ctx 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())
+	}
+
+	//Wait for all operations to complete
+	select {
+	case <-txn.opsFlow:
+	case <-time.After(txn.opTimeout):
+		return newError(TimeoutError, "There are some operations that are not completed after the timeout.")
+	}
+	//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))
+	} else {
+		if err.(*Error).Result() == TransactionNoFoundError || err.(*Error).Result() == InvalidStatus {
+			atomic.StoreInt32((*int32)(&txn.state), int32(TxnError))
+		} else {
+			txn.opsFlow <- true
+		}
+	}
+	return err
+}
+
+func (txn *transaction) registerSendOrAckOp() error {
+	if atomic.AddInt32(&txn.opsCount, 1) == 1 {
+		//There are new operations that not completed
+		select {
+		case <-txn.opsFlow:
+			return nil
+		case <-time.After(txn.opTimeout):
+			if _, err := txn.checkIfOpen(); err != nil {
+				return err
+			}
+			return newError(TimeoutError, "Failed to get the semaphore to register the send/ack operation")
+		}
+	}
+	return nil
+}
+
+func (txn *transaction) endSendOrAckOp(err error) {
+	if err != nil {
+		atomic.StoreInt32((*int32)(&txn.state), int32(TxnError))
+	}
+	if atomic.AddInt32(&txn.opsCount, -1) == 0 {
+		//This means there are not pending send/ack requests
+		txn.opsFlow <- true
+	}
+}
+
+func (txn *transaction) registerProducerTopic(topic string) error {
+	isOpen, err := txn.checkIfOpen()
+	if !isOpen {
+		return err
+	}
+	_, ok := txn.registerPartitions[topic]
+	if !ok {
+		txn.Lock()
+		defer txn.Unlock()
+		if _, ok = txn.registerPartitions[topic]; !ok {
+			err := txn.tcClient.addPublishPartitionToTxn(&txn.txnID, []string{topic})
+			if err != nil {
+				return err
+			}
+			txn.registerPartitions[topic] = true
+		}
+	}
+	return nil
+}
+
+func (txn *transaction) registerAckTopic(topic string, subName string) error {
+	isOpen, err := txn.checkIfOpen()
+	if !isOpen {
+		return err
+	}
+	sub := subscription{
+		topic:        topic,
+		subscription: subName,
+	}
+	_, ok := txn.registerAckSubscriptions[sub]
+	if !ok {
+		txn.Lock()
+		defer txn.Unlock()
+		if _, ok = txn.registerAckSubscriptions[sub]; !ok {
+			err := txn.tcClient.addSubscriptionToTxn(&txn.txnID, topic, subName)
+			if err != nil {
+				return err
+			}
+			txn.registerAckSubscriptions[sub] = true
+		}
+	}
+	return nil
+}
+
+func (txn *transaction) GetTxnID() TxnID {
+	return txn.txnID
+}
+
+func (txn *transaction) checkIfOpen() (bool, error) {
+	if txn.state == TxnOpen {
+		return true, nil
+	}
+	return false, newError(InvalidStatus, "Expect transaction state is TxnOpen but "+txn.state.string())
+}
+
+func (state TxnState) string() string {
+	switch state {
+	case TxnOpen:
+		return "TxnOpen"
+	case TxnCommitting:
+		return "TxnCommitting"
+	case TxnAborting:
+		return "TxnAborting"
+	case TxnCommitted:
+		return "TxnCommitted"
+	case TxnAborted:
+		return "TxnAborted"
+	case TxnTimeout:
+		return "TxnTimeout"
+	default:
+		return "Unknown"
+	}
+}


Test patch:
diff --git a/pulsar/ack_grouping_tracker_test.go b/pulsar/ack_grouping_tracker_test.go
index e7a6725..0a794f6 100644
--- a/pulsar/ack_grouping_tracker_test.go
+++ b/pulsar/ack_grouping_tracker_test.go
@@ -19,11 +19,13 @@ package pulsar
 
 import (
 	"fmt"
+	"sort"
 	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
 
+	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -45,7 +47,8 @@ func TestNoCacheTracker(t *testing.T) {
 				ledgerID1 := int64(-1)
 				tracker := newAckGroupingTracker(&option,
 					func(id MessageID) { ledgerID0 = id.LedgerID() },
-					func(id MessageID) { ledgerID1 = id.LedgerID() })
+					func(id MessageID) { ledgerID1 = id.LedgerID() },
+					nil)
 
 				tracker.add(&messageID{ledgerID: 1})
 				assert.Equal(t, atomic.LoadInt64(&ledgerID0), int64(1))
@@ -61,10 +64,12 @@ type mockAcker struct {
 	cumulativeLedgerID int64
 }
 
-func (a *mockAcker) ack(id MessageID) {
+func (a *mockAcker) ack(ids []*pb.MessageIdData) {
 	defer a.Unlock()
 	a.Lock()
-	a.ledgerIDs = append(a.ledgerIDs, id.LedgerID())
+	for _, id := range ids {
+		a.ledgerIDs = append(a.ledgerIDs, int64(*id.LedgerId))
+	}
 }
 
 func (a *mockAcker) ackCumulative(id MessageID) {
@@ -74,6 +79,8 @@ func (a *mockAcker) ackCumulative(id MessageID) {
 func (a *mockAcker) getLedgerIDs() []int64 {
 	defer a.Unlock()
 	a.Lock()
+
+	sort.Slice(a.ledgerIDs, func(i, j int) bool { return a.ledgerIDs[i] < a.ledgerIDs[j] })
 	return a.ledgerIDs
 }
 
@@ -88,8 +95,8 @@ func (a *mockAcker) reset() {
 
 func TestCachedTracker(t *testing.T) {
 	var acker mockAcker
-	tracker := newAckGroupingTracker(&AckGroupingOptions{MaxSize: 3, MaxTime: 0},
-		func(id MessageID) { acker.ack(id) }, func(id MessageID) { acker.ackCumulative(id) })
+	tracker := newAckGroupingTracker(&AckGroupingOptions{MaxSize: 3, MaxTime: 0}, nil,
+		func(id MessageID) { acker.ackCumulative(id) }, func(ids []*pb.MessageIdData) { acker.ack(ids) })
 
 	tracker.add(&messageID{ledgerID: 1})
 	tracker.add(&messageID{ledgerID: 2})
@@ -126,7 +133,8 @@ func TestCachedTracker(t *testing.T) {
 func TestTimedTrackerIndividualAck(t *testing.T) {
 	var acker mockAcker
 	// MaxSize: 1000, MaxTime: 100ms
-	tracker := newAckGroupingTracker(nil, func(id MessageID) { acker.ack(id) }, nil)
+	tracker := newAckGroupingTracker(nil, nil,
+		func(id MessageID) { acker.ackCumulative(id) }, func(ids []*pb.MessageIdData) { acker.ack(ids) })
 
 	expected := make([]int64, 0)
 	for i := 0; i < 999; i++ {
@@ -161,7 +169,7 @@ func TestTimedTrackerIndividualAck(t *testing.T) {
 func TestTimedTrackerCumulativeAck(t *testing.T) {
 	var acker mockAcker
 	// MaxTime is 100ms
-	tracker := newAckGroupingTracker(nil, nil, func(id MessageID) { acker.ackCumulative(id) })
+	tracker := newAckGroupingTracker(nil, nil, func(id MessageID) { acker.ackCumulative(id) }, nil)
 
 	// case 1: flush because of the timeout
 	tracker.addCumulative(&messageID{ledgerID: 1})
@@ -182,7 +190,8 @@ func TestTimedTrackerCumulativeAck(t *testing.T) {
 }
 
 func TestTimedTrackerIsDuplicate(t *testing.T) {
-	tracker := newAckGroupingTracker(nil, func(id MessageID) {}, func(id MessageID) {})
+	tracker := newAckGroupingTracker(nil, func(id MessageID) {}, func(id MessageID) {},
+		func(id []*pb.MessageIdData) {})
 
 	tracker.add(&messageID{batchIdx: 0, batchSize: 3})
 	tracker.add(&messageID{batchIdx: 2, batchSize: 3})
@@ -195,3 +204,40 @@ func TestTimedTrackerIsDuplicate(t *testing.T) {
 	assert.False(t, tracker.isDuplicate(&messageID{batchIdx: 1, batchSize: 3}))
 	assert.False(t, tracker.isDuplicate(&messageID{batchIdx: 2, batchSize: 3}))
 }
+
+func TestDuplicateAfterClose(t *testing.T) {
+	var acker mockAcker
+	tracker := newAckGroupingTracker(&AckGroupingOptions{MaxSize: 3, MaxTime: 0}, nil,
+		func(id MessageID) { acker.ackCumulative(id) }, func(ids []*pb.MessageIdData) { acker.ack(ids) })
+
+	tracker.add(&messageID{ledgerID: 1})
+	assert.True(t, tracker.isDuplicate(&messageID{ledgerID: 1}))
+
+	tracker.close()
+	assert.False(t, tracker.isDuplicate(&messageID{ledgerID: 1}))
+}
+
+func TestTrackerPendingAcks(t *testing.T) {
+	m := make(map[uint64][]int64)
+	tracker := newAckGroupingTracker(&AckGroupingOptions{MaxSize: 3, MaxTime: 0}, nil, nil,
+		func(ids []*pb.MessageIdData) {
+			for _, id := range ids {
+				m[*id.LedgerId] = id.AckSet
+			}
+		})
+	tracker.add(&messageID{ledgerID: 0, batchIdx: 0, batchSize: 30})
+	for i := 0; i < 10; i++ {
+		tracker.add(&messageID{ledgerID: 1, batchIdx: int32(i), batchSize: 10})
+	}
+	assert.Equal(t, 0, len(m)) // the number of entries is 2, so it's not flushed
+	tracker.flush()
+	assert.Equal(t, 2, len(m))
+
+	ackSet, found := m[0]
+	assert.True(t, found)
+	assert.Greater(t, len(ackSet), 0)
+
+	ackSet, found = m[1]
+	assert.True(t, found)
+	assert.Equal(t, 0, len(ackSet)) // all messages in the batch are acknowledged
+}
diff --git a/pulsar/consumer_partition_test.go b/pulsar/consumer_partition_test.go
index 16c4399..21280fb 100644
--- a/pulsar/consumer_partition_test.go
+++ b/pulsar/consumer_partition_test.go
@@ -37,8 +37,9 @@ func TestSingleMessageIDNoAckTracker(t *testing.T) {
 		metrics:              newTestMetrics(),
 		decryptor:            crypto.NewNoopDecryptor(),
 	}
+	pc.availablePermits = &availablePermits{pc: &pc}
 	pc.ackGroupingTracker = newAckGroupingTracker(&AckGroupingOptions{MaxSize: 0},
-		func(id MessageID) { pc.sendIndividualAck(id) }, nil)
+		func(id MessageID) { pc.sendIndividualAck(id) }, nil, nil)
 
 	headersAndPayload := internal.NewBufferWrapper(rawCompatSingleMessage)
 	if err := pc.MessageReceived(nil, headersAndPayload); err != nil {
@@ -75,8 +76,9 @@ func TestBatchMessageIDNoAckTracker(t *testing.T) {
 		metrics:              newTestMetrics(),
 		decryptor:            crypto.NewNoopDecryptor(),
 	}
+	pc.availablePermits = &availablePermits{pc: &pc}
 	pc.ackGroupingTracker = newAckGroupingTracker(&AckGroupingOptions{MaxSize: 0},
-		func(id MessageID) { pc.sendIndividualAck(id) }, nil)
+		func(id MessageID) { pc.sendIndividualAck(id) }, nil, nil)
 
 	headersAndPayload := internal.NewBufferWrapper(rawBatchMessage1)
 	if err := pc.MessageReceived(nil, headersAndPayload); err != nil {
@@ -110,8 +112,9 @@ func TestBatchMessageIDWithAckTracker(t *testing.T) {
 		metrics:              newTestMetrics(),
 		decryptor:            crypto.NewNoopDecryptor(),
 	}
+	pc.availablePermits = &availablePermits{pc: &pc}
 	pc.ackGroupingTracker = newAckGroupingTracker(&AckGroupingOptions{MaxSize: 0},
-		func(id MessageID) { pc.sendIndividualAck(id) }, nil)
+		func(id MessageID) { pc.sendIndividualAck(id) }, nil, nil)
 
 	headersAndPayload := internal.NewBufferWrapper(rawBatchMessage10)
 	if err := pc.MessageReceived(nil, headersAndPayload); err != nil {
diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
index 21fa7d0..521e576 100644
--- a/pulsar/consumer_test.go
+++ b/pulsar/consumer_test.go
@@ -3988,3 +3988,329 @@ func runBatchIndexAckTest(t *testing.T, ackWithResponse bool, cumulative bool, o
 
 	client.Close()
 }
+
+func TestConsumerWithAutoScaledQueueReceive(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topic := newTopicName()
+
+	// create consumer
+	c, err := client.Subscribe(ConsumerOptions{
+		Topic:                             topic,
+		SubscriptionName:                  "my-sub",
+		Type:                              Exclusive,
+		ReceiverQueueSize:                 3,
+		EnableAutoScaledReceiverQueueSize: true,
+	})
+	assert.Nil(t, err)
+	pc := c.(*consumer).consumers[0]
+	assert.Equal(t, int32(1), pc.currentQueueSize.Load())
+	defer c.Close()
+
+	// create p
+	p, err := client.CreateProducer(ProducerOptions{
+		Topic:           topic,
+		DisableBatching: false,
+	})
+	assert.Nil(t, err)
+	defer p.Close()
+
+	// send message, it will update scaleReceiverQueueHint from false to true
+	_, err = p.Send(context.Background(), &ProducerMessage{
+		Payload: []byte("hello"),
+	})
+	assert.NoError(t, err)
+
+	// this will trigger receiver queue size expanding to 2 because we have prefetched 1 message >= currentSize 1.
+	_, err = c.Receive(context.Background())
+	assert.Nil(t, err)
+
+	// currentQueueSize should be doubled in size
+	retryAssert(t, 5, 200, func() {}, func(t assert.TestingT) bool {
+		return assert.Equal(t, 2, int(pc.currentQueueSize.Load()))
+	})
+
+	for i := 0; i < 5; i++ {
+		_, err = p.Send(context.Background(), &ProducerMessage{
+			Payload: []byte("hello"),
+		})
+		assert.NoError(t, err)
+
+		// waiting for prefetched message passing from queueCh to messageCh
+		retryAssert(t, 5, 200, func() {}, func(t assert.TestingT) bool {
+			return assert.Equal(t, 1, len(pc.messageCh))
+		})
+
+		_, err = p.Send(context.Background(), &ProducerMessage{
+			Payload: []byte("hello"),
+		})
+		assert.NoError(t, err)
+
+		// wait all the messages has been prefetched
+		_, err = c.Receive(context.Background())
+		assert.Nil(t, err)
+		_, err = c.Receive(context.Background())
+		assert.Nil(t, err)
+		// this will not trigger receiver queue size expanding because we have prefetched 2 message < currentSize 4.
+		assert.Equal(t, int32(2), pc.currentQueueSize.Load())
+	}
+
+	for i := 0; i < 5; i++ {
+		p.SendAsync(
+			context.Background(),
+			&ProducerMessage{Payload: []byte("hello")},
+			func(id MessageID, producerMessage *ProducerMessage, err error) {
+			},
+		)
+	}
+
+	retryAssert(t, 3, 300, func() {}, func(t assert.TestingT) bool {
+		return assert.Equal(t, 3, int(pc.currentQueueSize.Load()))
+	})
+}
+
+func TestConsumerBatchIndexAckDisabled(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topic := newTopicName()
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:            topic,
+		SubscriptionName: "my-sub",
+	})
+	assert.Nil(t, err)
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic: topic,
+	})
+	assert.Nil(t, err)
+
+	for i := 0; i < 5; i++ {
+		producer.SendAsync(context.Background(), &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("msg-%d", i)),
+		}, nil)
+	}
+	for i := 0; i < 5; i++ {
+		message, err := consumer.Receive(context.Background())
+		assert.Nil(t, err)
+		consumer.Ack(message)
+	}
+	consumer.Close()
+	consumer, err = client.Subscribe(ConsumerOptions{
+		Topic:            topic,
+		SubscriptionName: "my-sub",
+	})
+	assert.Nil(t, err)
+	producer.Send(context.Background(), &ProducerMessage{Payload: []byte("done")})
+	message, err := consumer.Receive(context.Background())
+	assert.Nil(t, err)
+	assert.Equal(t, []byte("done"), message.Payload())
+}
+
+func TestConsumerMemoryLimit(t *testing.T) {
+	// Create client 1 without memory limit
+	cli1, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer cli1.Close()
+
+	// Create client 1 with memory limit
+	cli2, err := NewClient(ClientOptions{
+		URL:              lookupURL,
+		MemoryLimitBytes: 10 * 1024,
+	})
+
+	assert.Nil(t, err)
+	defer cli2.Close()
+
+	topic := newTopicName()
+
+	// Use client 1 to create producer p1
+	p1, err := cli1.CreateProducer(ProducerOptions{
+		Topic:           topic,
+		DisableBatching: false,
+	})
+	assert.Nil(t, err)
+	defer p1.Close()
+
+	// Use mem-limited client 2 to create consumer c1
+	c1, err := cli2.Subscribe(ConsumerOptions{
+		Topic:                             topic,
+		SubscriptionName:                  "my-sub-1",
+		Type:                              Exclusive,
+		EnableAutoScaledReceiverQueueSize: true,
+	})
+	assert.Nil(t, err)
+	defer c1.Close()
+	pc1 := c1.(*consumer).consumers[0]
+
+	// Fill up the messageCh of c1
+	for i := 0; i < 10; i++ {
+		p1.SendAsync(
+			context.Background(),
+			&ProducerMessage{Payload: createTestMessagePayload(1)},
+			func(id MessageID, producerMessage *ProducerMessage, err error) {
+			},
+		)
+	}
+
+	retryAssert(t, 5, 200, func() {}, func(t assert.TestingT) bool {
+		return assert.Equal(t, 10, len(pc1.messageCh))
+	})
+
+	// Get current receiver queue size of c1
+	prevQueueSize := pc1.currentQueueSize.Load()
+
+	// Make the client 1 exceed the memory limit
+	_, err = p1.Send(context.Background(), &ProducerMessage{
+		Payload: createTestMessagePayload(10*1024 + 1),
+	})
+	assert.NoError(t, err)
+
+	// c1 should shrink it's receiver queue size
+	retryAssert(t, 5, 200, func() {}, func(t assert.TestingT) bool {
+		return assert.Equal(t, prevQueueSize/2, pc1.currentQueueSize.Load())
+	})
+
+	// Use mem-limited client 2 to create consumer c2
+	c2, err := cli2.Subscribe(ConsumerOptions{
+		Topic:                             topic,
+		SubscriptionName:                  "my-sub-2",
+		Type:                              Exclusive,
+		SubscriptionInitialPosition:       SubscriptionPositionEarliest,
+		EnableAutoScaledReceiverQueueSize: true,
+	})
+	assert.Nil(t, err)
+	defer c2.Close()
+	pc2 := c2.(*consumer).consumers[0]
+
+	// Try to induce c2 receiver queue size expansion
+	for i := 0; i < 10; i++ {
+		p1.SendAsync(
+			context.Background(),
+			&ProducerMessage{Payload: createTestMessagePayload(1)},
+			func(id MessageID, producerMessage *ProducerMessage, err error) {
+			},
+		)
+	}
+
+	retryAssert(t, 5, 200, func() {}, func(t assert.TestingT) bool {
+		return assert.Equal(t, 10, len(pc1.messageCh))
+	})
+
+	// c2 receiver queue size should not expansion because client 1 has exceeded the memory limit
+	assert.Equal(t, 1, int(pc2.currentQueueSize.Load()))
+
+	// Use mem-limited client 2 to create producer p2
+	p2, err := cli2.CreateProducer(ProducerOptions{
+		Topic:                   topic,
+		DisableBatching:         false,
+		DisableBlockIfQueueFull: true,
+	})
+	assert.Nil(t, err)
+	defer p2.Close()
+
+	_, err = p2.Send(context.Background(), &ProducerMessage{
+		Payload: createTestMessagePayload(1),
+	})
+	// Producer can't send message
+	assert.Equal(t, true, errors.Is(err, errMemoryBufferIsFull))
+}
+
+func TestMultiConsumerMemoryLimit(t *testing.T) {
+	// Create client 1 without memory limit
+	cli1, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer cli1.Close()
+
+	// Create client 1 with memory limit
+	cli2, err := NewClient(ClientOptions{
+		URL:              lookupURL,
+		MemoryLimitBytes: 10 * 1024,
+	})
+
+	assert.Nil(t, err)
+	defer cli2.Close()
+
+	topic := newTopicName()
+
+	// Use client 1 to create producer p1
+	p1, err := cli1.CreateProducer(ProducerOptions{
+		Topic:           topic,
+		DisableBatching: false,
+	})
+	assert.Nil(t, err)
+	defer p1.Close()
+
+	// Use mem-limited client 2 to create consumer c1
+	c1, err := cli2.Subscribe(ConsumerOptions{
+		Topic:                             topic,
+		SubscriptionName:                  "my-sub-1",
+		Type:                              Exclusive,
+		EnableAutoScaledReceiverQueueSize: true,
+	})
+	assert.Nil(t, err)
+	defer c1.Close()
+	pc1 := c1.(*consumer).consumers[0]
+
+	// Use mem-limited client 2 to create consumer c1
+	c2, err := cli2.Subscribe(ConsumerOptions{
+		Topic:                             topic,
+		SubscriptionName:                  "my-sub-2",
+		Type:                              Exclusive,
+		EnableAutoScaledReceiverQueueSize: true,
+	})
+	assert.Nil(t, err)
+	defer c2.Close()
+	pc2 := c2.(*consumer).consumers[0]
+
+	// Fill up the messageCh of c1 nad c2
+	for i := 0; i < 10; i++ {
+		p1.SendAsync(
+			context.Background(),
+			&ProducerMessage{Payload: createTestMessagePayload(1)},
+			func(id MessageID, producerMessage *ProducerMessage, err error) {
+			},
+		)
+	}
+
+	retryAssert(t, 5, 200, func() {}, func(t assert.TestingT) bool {
+		return assert.Equal(t, 10, len(pc1.messageCh))
+	})
+
+	retryAssert(t, 5, 200, func() {}, func(t assert.TestingT) bool {
+		return assert.Equal(t, 10, len(pc2.messageCh))
+	})
+
+	// Get current receiver queue size of c1 and c2
+	pc1PrevQueueSize := pc1.currentQueueSize.Load()
+	pc2PrevQueueSize := pc2.currentQueueSize.Load()
+
+	// Make the client 1 exceed the memory limit
+	_, err = p1.Send(context.Background(), &ProducerMessage{
+		Payload: createTestMessagePayload(10*1024 + 1),
+	})
+	assert.NoError(t, err)
+
+	// c1 should shrink it's receiver queue size
+	retryAssert(t, 5, 200, func() {}, func(t assert.TestingT) bool {
+		return assert.Equal(t, pc1PrevQueueSize/2, pc1.currentQueueSize.Load())
+	})
+
+	// c2 should shrink it's receiver queue size too
+	retryAssert(t, 5, 200, func() {}, func(t assert.TestingT) bool {
+		return assert.Equal(t, pc2PrevQueueSize/2, pc2.currentQueueSize.Load())
+	})
+}
diff --git a/pulsar/helper_for_test.go b/pulsar/helper_for_test.go
index eda2d72..7bbf66e 100644
--- a/pulsar/helper_for_test.go
+++ b/pulsar/helper_for_test.go
@@ -177,9 +177,14 @@ func retryAssert(t assert.TestingT, times int, milliseconds int, update func(),
 	for i := 0; i < times; i++ {
 		time.Sleep(time.Duration(milliseconds) * time.Millisecond)
 		update()
-		if assert(nil) {
+		if assert(fakeAssertT{}) {
 			break
 		}
 	}
 	assert(t)
 }
+
+type fakeAssertT struct{}
+
+func (fa fakeAssertT) Errorf(format string, args ...interface{}) {
+}
diff --git a/pulsar/internal/memory_limit_controller_test.go b/pulsar/internal/memory_limit_controller_test.go
index a62c6e6..81623ec 100644
--- a/pulsar/internal/memory_limit_controller_test.go
+++ b/pulsar/internal/memory_limit_controller_test.go
@@ -28,7 +28,7 @@ import (
 
 func TestLimit(t *testing.T) {
 
-	mlc := NewMemoryLimitController(100)
+	mlc := NewMemoryLimitController(100, 1.0)
 
 	for i := 0; i < 101; i++ {
 		assert.True(t, mlc.TryReserveMemory(1))
@@ -57,7 +57,7 @@ func TestLimit(t *testing.T) {
 }
 
 func TestDisableLimit(t *testing.T) {
-	mlc := NewMemoryLimitController(-1)
+	mlc := NewMemoryLimitController(-1, 1.0)
 	assert.True(t, mlc.TryReserveMemory(1000000))
 	assert.True(t, mlc.ReserveMemory(context.Background(), 1000000))
 	mlc.ReleaseMemory(1000000)
@@ -65,7 +65,7 @@ func TestDisableLimit(t *testing.T) {
 }
 
 func TestMultiGoroutineTryReserveMem(t *testing.T) {
-	mlc := NewMemoryLimitController(10000)
+	mlc := NewMemoryLimitController(10000, 1.0)
 
 	// Multi goroutine try reserve memory.
 	wg := sync.WaitGroup{}
@@ -87,7 +87,7 @@ func TestMultiGoroutineTryReserveMem(t *testing.T) {
 }
 
 func TestReserveWithContext(t *testing.T) {
-	mlc := NewMemoryLimitController(100)
+	mlc := NewMemoryLimitController(100, 1.0)
 	assert.True(t, mlc.TryReserveMemory(101))
 	gorNum := 10
 
@@ -120,7 +120,7 @@ func TestReserveWithContext(t *testing.T) {
 }
 
 func TestBlocking(t *testing.T) {
-	mlc := NewMemoryLimitController(100)
+	mlc := NewMemoryLimitController(100, 1.0)
 	assert.True(t, mlc.TryReserveMemory(101))
 	assert.Equal(t, int64(101), mlc.CurrentUsage())
 	assert.InDelta(t, 1.01, mlc.CurrentUsagePercent(), 0.000001)
@@ -146,7 +146,7 @@ func TestBlocking(t *testing.T) {
 }
 
 func TestStepRelease(t *testing.T) {
-	mlc := NewMemoryLimitController(100)
+	mlc := NewMemoryLimitController(100, 1.0)
 	assert.True(t, mlc.TryReserveMemory(101))
 	assert.Equal(t, int64(101), mlc.CurrentUsage())
 	assert.InDelta(t, 1.01, mlc.CurrentUsagePercent(), 0.000001)
@@ -169,6 +169,57 @@ func TestStepRelease(t *testing.T) {
 	assert.Equal(t, int64(101), mlc.CurrentUsage())
 }
 
+func TestRegisterTrigger(t *testing.T) {
+	mlc := NewMemoryLimitController(100, 0.95)
+	triggeredResult1 := false
+	triggeredResult2 := false
+	finishCh := make(chan struct{}, 2)
+
+	mlc.RegisterTrigger(func() {
+		triggeredResult1 = true
+		finishCh <- struct{}{}
+	})
+
+	mlc.RegisterTrigger(func() {
+		triggeredResult2 = true
+		finishCh <- struct{}{}
+	})
+
+	mlc.TryReserveMemory(50)
+	timer := time.NewTimer(time.Millisecond * 500)
+	select {
+	case <-finishCh:
+		assert.Fail(t, "should not be triggered")
+	case <-timer.C:
+	}
+
+	mlc.TryReserveMemory(45)
+	timer.Reset(time.Millisecond * 500)
+	for i := 0; i < 2; i++ {
+		select {
+		case <-finishCh:
+		case <-timer.C:
+			assert.Fail(t, "trigger timeout")
+		}
+	}
+
+	assert.True(t, triggeredResult1)
+	assert.True(t, triggeredResult2)
+
+	triggeredResult2 = false
+	mlc.ReleaseMemory(1)
+	mlc.ForceReserveMemory(1)
+	timer.Reset(time.Millisecond * 500)
+	for i := 0; i < 2; i++ {
+		select {
+		case <-finishCh:
+		case <-timer.C:
+			assert.Fail(t, "trigger timeout")
+		}
+	}
+	assert.True(t, triggeredResult2)
+}
+
 func reserveMemory(mlc MemoryLimitController, ch chan int) {
 	mlc.ReserveMemory(context.Background(), 1)
 	ch <- 1
diff --git a/pulsar/reader_test.go b/pulsar/reader_test.go
index 3543187..0a1b2a1 100644
--- a/pulsar/reader_test.go
+++ b/pulsar/reader_test.go
@@ -70,6 +70,45 @@ func TestReaderConfigSubscribeName(t *testing.T) {
 	assert.NotNil(t, consumer)
 }
 
+func TestReaderConfigChunk(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+
+	r1, err := client.CreateReader(ReaderOptions{
+		Topic:                       "my-topic1",
+		StartMessageID:              EarliestMessageID(),
+		MaxPendingChunkedMessage:    50,
+		ExpireTimeOfIncompleteChunk: 30 * time.Second,
+		AutoAckIncompleteChunk:      true,
+	})
+	assert.Nil(t, err)
+	defer r1.Close()
+
+	// verify specified chunk options
+	pcOpts := r1.(*reader).pc.options
+	assert.Equal(t, 50, pcOpts.maxPendingChunkedMessage)
+	assert.Equal(t, 30*time.Second, pcOpts.expireTimeOfIncompleteChunk)
+	assert.True(t, pcOpts.autoAckIncompleteChunk)
+
+	r2, err := client.CreateReader(ReaderOptions{
+		Topic:          "my-topic2",
+		StartMessageID: EarliestMessageID(),
+	})
+	assert.Nil(t, err)
+	defer r2.Close()
+
+	// verify default chunk options
+	pcOpts = r2.(*reader).pc.options
+	assert.Equal(t, 100, pcOpts.maxPendingChunkedMessage)
+	assert.Equal(t, time.Minute, pcOpts.expireTimeOfIncompleteChunk)
+	assert.False(t, pcOpts.autoAckIncompleteChunk)
+}
+
 func TestReader(t *testing.T) {
 	client, err := NewClient(ClientOptions{
 		URL: lookupURL,
diff --git a/pulsar/transaction_test.go b/pulsar/transaction_test.go
index 14a652f..362e4d2 100644
--- a/pulsar/transaction_test.go
+++ b/pulsar/transaction_test.go
@@ -18,11 +18,15 @@
 package pulsar
 
 import (
-	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
-	"github.com/stretchr/testify/assert"
-
+	"context"
+	"errors"
+	"fmt"
 	"testing"
 	"time"
+
+	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestTCClient(t *testing.T) {
@@ -79,6 +83,126 @@ func TestTCClient(t *testing.T) {
 	defer client.Close()
 }
 
+//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.
+//	2. The internal API, registerSendOrAckOp and endSendOrAckOp
+//		1. Register 4 operation but only end 3 operations, the transaction can not be committed or aborted.
+//		2. Register 3 operation and end 3 operation the transaction can be committed and aborted.
+//		3. Register an operation and end the operation with an error,
+//		and then the state of the transaction should be replaced to errored.
+//	3. The internal API, registerAckTopic and registerProducerTopic
+//		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) {
+	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.
+	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))
+	txn1.state = 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)
+	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.opsFlow <- true
+	err = txn2.Abort(context.Background())
+	assert.Equal(t, err.(*Error).Result(), TransactionNoFoundError)
+	assert.Equal(t, txn1.GetState(), TxnError)
+	err = txn2.registerSendOrAckOp()
+	assert.Equal(t, err.(*Error).Result(), InvalidStatus)
+	err = txn1.registerSendOrAckOp()
+	assert.Equal(t, err.(*Error).Result(), InvalidStatus)
+}
+
+// TestRegisterOpAndEndOp Test the internal API including the registerSendOrAckOp and endSendOrAckOp.
+func TestRegisterOpAndEndOp(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)
+	assert.Equal(t, res.(*Error).Result(), TimeoutError)
+	res = registerOpAndEndOp(t, tc, 4, 3, nil, false)
+	assert.Equal(t, res.(*Error).Result(), TimeoutError)
+
+	//2. Register 3 operation and end 3 operation the transaction can be committed and aborted.
+	res = registerOpAndEndOp(t, tc, 3, 3, nil, true)
+	assert.Nil(t, res)
+	res = registerOpAndEndOp(t, tc, 3, 3, nil, false)
+	assert.Nil(t, res)
+	//3. Register an operation and end the operation with an error,
+	// and then the state of the transaction should be replaced to errored.
+	res = registerOpAndEndOp(t, tc, 4, 4, errors.New(""), true)
+	assert.Equal(t, res.(*Error).Result(), InvalidStatus)
+	res = registerOpAndEndOp(t, tc, 4, 4, errors.New(""), false)
+	assert.Equal(t, res.(*Error).Result(), InvalidStatus)
+}
+
+// TestRegisterTopic Test the internal API, registerAckTopic and registerProducerTopic
+func TestRegisterTopic(t *testing.T) {
+	//1. Prepare: create PulsarClient and init transaction coordinator client.
+	topic := newTopicName()
+	sub := "my-sub"
+	tc, client := createTcClient(t)
+	//2. Prepare: create Topic and Subscription.
+	_, err := client.Subscribe(ConsumerOptions{
+		Topic:            topic,
+		SubscriptionName: sub,
+	})
+	assert.NoError(t, err)
+	txn := createTxn(tc, t)
+	//3. Create a topic and register topic and subscription.
+	err = txn.registerAckTopic(topic, sub)
+	require.Nil(t, err, "Failed to register ack topic.")
+	err = txn.registerProducerTopic(topic)
+	require.Nil(t, err, "Failed to register ack topic.")
+	//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{})
+	assert.NotNil(t, topics[topic])
+	assert.NotNil(t, subTopics[topic])
+	subs := subTopics[topic].(map[string]interface{})
+	assert.NotNil(t, subs[sub])
+}
+
+func registerOpAndEndOp(t *testing.T, tc *transactionCoordinatorClient, rp int, ep int, err error, commit bool) error {
+	txn := createTxn(tc, t)
+	for i := 0; i < rp; i++ {
+		err := txn.registerSendOrAckOp()
+		assert.Nil(t, err)
+	}
+	for i := 0; i < ep; i++ {
+		txn.endSendOrAckOp(err)
+	}
+	if commit {
+		err = txn.Commit(context.Background())
+	} else {
+		err = txn.Abort(context.Background())
+	}
+	return err
+}
+
+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)
+}
+
 // createTcClient Create a transaction coordinator client to send request
 func createTcClient(t *testing.T) (*transactionCoordinatorClient, *client) {
 	c, err := NewClient(ClientOptions{
@@ -86,13 +210,10 @@ func createTcClient(t *testing.T) (*transactionCoordinatorClient, *client) {
 		TLSTrustCertsFilePath: caCertsPath,
 		Authentication:        NewAuthenticationTLS(tlsClientCertPath, tlsClientKeyPath),
 	})
-	if err != nil {
-		t.Fatalf("Failed to create client due to %s", err.Error())
-	}
+	require.Nil(t, err, "Failed to create client.")
 	tcClient := newTransactionCoordinatorClientImpl(c.(*client))
-	if err = tcClient.start(); err != nil {
-		t.Fatalf("Failed to start transaction coordinator due to %s", err.Error())
-	}
+	err = tcClient.start()
+	require.Nil(t, err, "Failed to start transaction coordinator.")
 
 	return tcClient, c.(*client)
 }