Repository: apache/pulsar-client-go
Pull request #960 (2023-02-21 07:19:17)

[improve] Athenz auth provider supports Copper Argos
### Motivation

As with the Java client (cf. https://github.com/apache/pulsar/pull/19445), make the Athenz auth provider for the Go client support [Copper Argos](https://github.com/AthenZ/athenz/blob/master/docs/copper_argos.md).

### Modifications

Added new parameters `x509CertChain` and `caCert` to the Athenz auth provider. If the parameter `x509CertChain` is passed, the auth provider assumes that Copper Argos is used and presents the X.509 certificate to obtain a role token from ZTS. Existing users will not be affected by this change.

### Verifying this change

- [x] Make sure that the change passes the CI checks.



Patch:
diff --git a/Makefile b/Makefile
index 3245434..231c05e 100644
--- a/Makefile
+++ b/Makefile
@@ -18,7 +18,7 @@
 #
 
 IMAGE_NAME = pulsar-client-go-test:latest
-PULSAR_VERSION ?= 2.8.3
+PULSAR_VERSION ?= 2.10.3
 PULSAR_IMAGE = apachepulsar/pulsar:$(PULSAR_VERSION)
 GO_VERSION ?= 1.18
 GOLANG_IMAGE = golang:$(GO_VERSION)
diff --git a/distribution/LICENSE.bin.txt b/distribution/LICENSE.bin.txt
index e37200f..79e2c52 100644
--- a/distribution/LICENSE.bin.txt
+++ b/distribution/LICENSE.bin.txt
@@ -218,9 +218,6 @@ MIT License
   * github.com/kr/pretty v0.2.0 -- distribution/license/LICENSE-pretty.txt
   * github.com/sirupsen/logrus v1.4.1 -- distribution/license/LICENSE-logrus.txt
 
-GNU GENERAL PUBLIC LICENSE Version 3
-  * github.com/beefsack/go-rate v0.0.0-20180408011153-efa7637bb9b6 -- distribution/license/LICENSE-go-rate.txt
-
 BSD License
   * github.com/DataDog/zstd v1.4.6-0.20210211175136-c6db21d202f4 -- distribution/license/LICENSE-zstd.txt
   * github.com/klauspost/compress v1.9.2 -- distribution/license/LICENSE-compress.txt
diff --git a/perf/perf-consumer.go b/perf/perf-consumer.go
index 7fb8aab..6b6e411 100644
--- a/perf/perf-consumer.go
+++ b/perf/perf-consumer.go
@@ -31,9 +31,10 @@ import (
 
 // ConsumeArgs define the parameters required by consume
 type ConsumeArgs struct {
-	Topic             string
-	SubscriptionName  string
-	ReceiverQueueSize int
+	Topic               string
+	SubscriptionName    string
+	ReceiverQueueSize   int
+	EnableBatchIndexAck bool
 }
 
 func newConsumerCommand() *cobra.Command {
@@ -55,6 +56,7 @@ func newConsumerCommand() *cobra.Command {
 	flags := cmd.Flags()
 	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")
 
 	return cmd
 }
@@ -74,8 +76,9 @@ func consume(consumeArgs *ConsumeArgs, stop <-chan struct{}) {
 	defer client.Close()
 
 	consumer, err := client.Subscribe(pulsar.ConsumerOptions{
-		Topic:            consumeArgs.Topic,
-		SubscriptionName: consumeArgs.SubscriptionName,
+		Topic:                          consumeArgs.Topic,
+		SubscriptionName:               consumeArgs.SubscriptionName,
+		EnableBatchIndexAcknowledgment: consumeArgs.EnableBatchIndexAck,
 	})
 
 	if err != nil {
diff --git a/perf/perf-producer.go b/perf/perf-producer.go
index 0ee0083..9d062ae 100644
--- a/perf/perf-producer.go
+++ b/perf/perf-producer.go
@@ -32,12 +32,13 @@ import (
 
 // ProduceArgs define the parameters required by produce
 type ProduceArgs struct {
-	Topic              string
-	Rate               int
-	BatchingTimeMillis int
-	BatchingMaxSize    int
-	MessageSize        int
-	ProducerQueueSize  int
+	Topic               string
+	Rate                int
+	BatchingTimeMillis  int
+	BatchingMaxSize     uint
+	BatchingNumMessages uint
+	MessageSize         int
+	ProducerQueueSize   int
 }
 
 func newProducerCommand() *cobra.Command {
@@ -62,8 +63,10 @@ func newProducerCommand() *cobra.Command {
 		"Publish rate. Set to 0 to go unthrottled")
 	flags.IntVarP(&produceArgs.BatchingTimeMillis, "batching-time", "b", 1,
 		"Batching grouping time in millis")
-	flags.IntVarP(&produceArgs.BatchingMaxSize, "batching-max-size", "", 128,
+	flags.UintVar(&produceArgs.BatchingMaxSize, "batching-max-size", 128,
 		"Max size of a batch (in KB)")
+	flags.UintVar(&produceArgs.BatchingNumMessages, "batching-num-messages", 1000,
+		"Maximum number of messages permitted in a batch")
 	flags.IntVarP(&produceArgs.MessageSize, "size", "s", 1024,
 		"Message size")
 	flags.IntVarP(&produceArgs.ProducerQueueSize, "queue-size", "q", 1000,
@@ -88,7 +91,8 @@ func produce(produceArgs *ProduceArgs, stop <-chan struct{}) {
 		Topic:                   produceArgs.Topic,
 		MaxPendingMessages:      produceArgs.ProducerQueueSize,
 		BatchingMaxPublishDelay: time.Millisecond * time.Duration(produceArgs.BatchingTimeMillis),
-		BatchingMaxSize:         uint(produceArgs.BatchingMaxSize * 1024),
+		BatchingMaxSize:         produceArgs.BatchingMaxSize * 1024,
+		BatchingMaxMessages:     produceArgs.BatchingNumMessages,
 	})
 	if err != nil {
 		log.Fatal(err)
diff --git a/pulsar/ack_grouping_tracker.go b/pulsar/ack_grouping_tracker.go
new file mode 100644
index 0000000..dbc70f5
--- /dev/null
+++ b/pulsar/ack_grouping_tracker.go
@@ -0,0 +1,298 @@
+// 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 (
+	"time"
+
+	"github.com/bits-and-blooms/bitset"
+)
+
+type ackGroupingTracker interface {
+	add(id MessageID)
+
+	addCumulative(id MessageID)
+
+	isDuplicate(id MessageID) bool
+
+	flush()
+
+	flushAndClean()
+
+	close()
+}
+
+type ackFlushType int
+
+const (
+	flushOnly ackFlushType = iota
+	flushAndClean
+	flushAndClose
+)
+
+func newAckGroupingTracker(options *AckGroupingOptions,
+	ackIndividual func(id MessageID),
+	ackCumulative func(id MessageID)) ackGroupingTracker {
+	if options == nil {
+		options = &AckGroupingOptions{
+			MaxSize: 1000,
+			MaxTime: 100 * time.Millisecond,
+		}
+	}
+
+	if options.MaxSize <= 1 {
+		return &immediateAckGroupingTracker{
+			ackIndividual: ackIndividual,
+			ackCumulative: ackCumulative,
+		}
+	}
+
+	c := &cachedAcks{
+		singleAcks:        make([]MessageID, options.MaxSize),
+		pendingAcks:       make(map[int64]*bitset.BitSet),
+		lastCumulativeAck: EarliestMessageID(),
+		ackIndividual:     ackIndividual,
+		ackCumulative:     ackCumulative,
+		ackList: func(ids []MessageID) {
+			// TODO: support ack a list of MessageIDs
+			for _, id := range ids {
+				ackIndividual(id)
+			}
+		},
+	}
+
+	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 {
+					return
+				}
+			}
+		}
+	}()
+	return t
+}
+
+type immediateAckGroupingTracker struct {
+	ackIndividual func(id MessageID)
+	ackCumulative func(id MessageID)
+}
+
+func (i *immediateAckGroupingTracker) add(id MessageID) {
+	i.ackIndividual(id)
+}
+
+func (i *immediateAckGroupingTracker) addCumulative(id MessageID) {
+	i.ackCumulative(id)
+}
+
+func (i *immediateAckGroupingTracker) isDuplicate(id MessageID) bool {
+	return false
+}
+
+func (i *immediateAckGroupingTracker) flush() {
+}
+
+func (i *immediateAckGroupingTracker) flushAndClean() {
+}
+
+func (i *immediateAckGroupingTracker) close() {
+}
+
+type cachedAcks struct {
+	singleAcks []MessageID
+	index      int
+
+	// Key is the hash code of the ledger id and the netry 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
+
+	lastCumulativeAck     MessageID
+	cumulativeAckRequired bool
+
+	ackIndividual func(id MessageID)
+	ackCumulative func(id MessageID)
+	ackList       func(ids []MessageID)
+}
+
+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))
+			}
+			t.pendingAcks[key] = ackSet
+		} else {
+			t.pendingAcks[key] = nil
+		}
+	}
+	if ackSet != nil {
+		ackSet.Clear(uint(id.BatchIdx()))
+	}
+	return t.index == len(t.singleAcks)
+}
+
+func (t *cachedAcks) tryUpdateLastCumulativeAck(id MessageID) {
+	if messageIDCompare(t.lastCumulativeAck, id) < 0 {
+		t.lastCumulativeAck = id
+		t.cumulativeAckRequired = true
+	}
+}
+
+func (t *cachedAcks) isDuplicate(id MessageID) bool {
+	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))
+		}
+		t.index = 0
+	}
+}
+
+func (t *cachedAcks) flushCumulativeAck() {
+	if t.cumulativeAckRequired {
+		t.ackCumulative(t.lastCumulativeAck)
+		t.cumulativeAckRequired = false
+	}
+}
+
+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) flush() {
+	t.flushCh <- flushOnly
+	<-t.waitFlushCh
+}
+
+func (t *timedAckGroupingTracker) flushAndClean() {
+	t.flushCh <- flushAndClean
+	<-t.waitFlushCh
+}
+
+func (t *timedAckGroupingTracker) close() {
+	t.flushCh <- flushAndClose
+	<-t.waitFlushCh
+}
diff --git a/pulsar/auth/athenz.go b/pulsar/auth/athenz.go
index f490be9..35b3cc4 100644
--- a/pulsar/auth/athenz.go
+++ b/pulsar/auth/athenz.go
@@ -32,28 +32,33 @@ import (
 )
 
 const (
-	minExpire            = 2 * time.Hour
-	maxExpire            = 24 * time.Hour
-	AthenzRoleAuthHeader = "Athenz-Role-Auth"
+	minExpire         = 2 * time.Hour
+	maxExpire         = 24 * time.Hour
+	defaultKeyID      = "0"
+	defaultRoleHeader = "Athenz-Role-Auth"
 )
 
 type athenzAuthProvider struct {
-	providerDomain     string
-	tenantDomain       string
-	tenantService      string
-	privateKey         string
-	keyID              string
-	principalHeader    string
-	ztsURL             string
-	tokenBuilder       zms.TokenBuilder
-	roleToken          zts.RoleToken
-	zmsNewTokenBuilder func(domain, name string, privateKeyPEM []byte, keyVersion string) (zms.TokenBuilder, error)
-	ztsNewRoleToken    func(tok zms.Token, domain string, opts zts.RoleTokenOptions) zts.RoleToken
+	providerDomain          string
+	tenantDomain            string
+	tenantService           string
+	privateKey              string
+	keyID                   string
+	x509CertChain           string
+	caCert                  string
+	principalHeader         string
+	roleHeader              string
+	ztsURL                  string
+	tokenBuilder            zms.TokenBuilder
+	roleToken               zts.RoleToken
+	zmsNewTokenBuilder      func(domain, name string, privateKeyPEM []byte, keyVersion string) (zms.TokenBuilder, error)
+	ztsNewRoleToken         func(tok zms.Token, domain string, opts zts.RoleTokenOptions) zts.RoleToken
+	ztsNewRoleTokenFromCert func(certFile, keyFile, domain string, opts zts.RoleTokenOptions) zts.RoleToken
 
 	T http.RoundTripper
 }
 
-type privateKeyURI struct {
+type parsedURI struct {
 	Scheme                   string
 	MediaTypeAndEncodingType string
 	Data                     string
@@ -67,7 +72,10 @@ func NewAuthenticationAthenzWithParams(params map[string]string) (Provider, erro
 		params["tenantService"],
 		params["privateKey"],
 		params["keyId"],
+		params["x509CertChain"],
+		params["caCert"],
 		params["principalHeader"],
+		params["roleHeader"],
 		params["ztsUrl"],
 	), nil
 }
@@ -78,68 +86,97 @@ func NewAuthenticationAthenz(
 	tenantService string,
 	privateKey string,
 	keyID string,
+	x509CertChain string,
+	caCert string,
 	principalHeader string,
+	roleHeader string,
 	ztsURL string) Provider {
-	var fixedKeyID string
-	if keyID == "" {
-		fixedKeyID = "0"
-	} else {
+	fixedKeyID := defaultKeyID
+	if keyID != "" {
 		fixedKeyID = keyID
 	}
+
+	fixedRoleHeader := defaultRoleHeader
+	if roleHeader != "" {
+		fixedRoleHeader = roleHeader
+	}
+
 	ztsNewRoleToken := func(tok zms.Token, domain string, opts zts.RoleTokenOptions) zts.RoleToken {
 		return zts.RoleToken(zts.NewRoleToken(tok, domain, opts))
 	}
 
+	ztsNewRoleTokenFromCert := func(certFile, keyFile, domain string, opts zts.RoleTokenOptions) zts.RoleToken {
+		return zts.RoleToken(zts.NewRoleTokenFromCert(certFile, keyFile, domain, opts))
+	}
+
 	return &athenzAuthProvider{
-		providerDomain:     providerDomain,
-		tenantDomain:       tenantDomain,
-		tenantService:      tenantService,
-		privateKey:         privateKey,
-		keyID:              fixedKeyID,
-		principalHeader:    principalHeader,
-		ztsURL:             strings.TrimSuffix(ztsURL, "/"),
-		zmsNewTokenBuilder: zms.NewTokenBuilder,
-		ztsNewRoleToken:    ztsNewRoleToken,
+		providerDomain:          providerDomain,
+		tenantDomain:            tenantDomain,
+		tenantService:           tenantService,
+		privateKey:              privateKey,
+		keyID:                   fixedKeyID,
+		x509CertChain:           x509CertChain,
+		caCert:                  caCert,
+		principalHeader:         principalHeader,
+		roleHeader:              fixedRoleHeader,
+		ztsURL:                  strings.TrimSuffix(ztsURL, "/"),
+		zmsNewTokenBuilder:      zms.NewTokenBuilder,
+		ztsNewRoleToken:         ztsNewRoleToken,
+		ztsNewRoleTokenFromCert: ztsNewRoleTokenFromCert,
 	}
 }
 
 func (p *athenzAuthProvider) Init() error {
-	uriSt := parseURI(p.privateKey)
-	var keyData []byte
+	if p.providerDomain == "" || p.privateKey == "" || p.ztsURL == "" {
+		return errors.New("missing required parameters")
+	}
 
-	if uriSt.Scheme == "data" {
-		if uriSt.MediaTypeAndEncodingType != "application/x-pem-file;base64" {
-			return errors.New("Unsupported mediaType or encodingType: " + uriSt.MediaTypeAndEncodingType)
+	var roleToken zts.RoleToken
+	opts := zts.RoleTokenOptions{
+		BaseZTSURL: p.ztsURL + "/zts/v1",
+		MinExpire:  minExpire,
+		MaxExpire:  maxExpire,
+		AuthHeader: p.principalHeader,
+	}
+
+	if p.x509CertChain != "" {
+		// use Copper Argos
+		certURISt := parseURI(p.x509CertChain)
+		keyURISt := parseURI(p.privateKey)
+
+		if certURISt.Scheme != "file" || keyURISt.Scheme != "file" {
+			return errors.New("x509CertChain and privateKey must be specified as file paths")
+		}
+
+		if p.caCert != "" {
+			caCertData, err := loadPEM(p.caCert)
+			if err != nil {
+				return err
+			}
+			opts.CACert = caCertData
 		}
-		key, err := base64.StdEncoding.DecodeString(uriSt.Data)
+
+		roleToken = p.ztsNewRoleTokenFromCert(certURISt.Path, keyURISt.Path, p.providerDomain, opts)
+	} else {
+		if p.tenantDomain == "" || p.tenantService == "" {
+			return errors.New("missing required parameters")
+		}
+
+		keyData, err := loadPEM(p.privateKey)
 		if err != nil {
 			return err
 		}
-		keyData = key
-	} else if uriSt.Scheme == "file" {
-		key, err := os.ReadFile(uriSt.Path)
+
+		tb, err := p.zmsNewTokenBuilder(p.tenantDomain, p.tenantService, keyData, p.keyID)
 		if err != nil {
 			return err
 		}
-		keyData = key
-	} else {
-		return errors.New("Unsupported URI Scheme: " + uriSt.Scheme)
-	}
+		p.tokenBuilder = tb
 
-	tb, err := p.zmsNewTokenBuilder(p.tenantDomain, p.tenantService, keyData, p.keyID)
-	if err != nil {
-		return err
+		roleToken = p.ztsNewRoleToken(p.tokenBuilder.Token(), p.providerDomain, opts)
 	}
-	p.tokenBuilder = tb
 
-	roleToken := p.ztsNewRoleToken(p.tokenBuilder.Token(), p.providerDomain, zts.RoleTokenOptions{
-		BaseZTSURL: p.ztsURL + "/zts/v1",
-		MinExpire:  minExpire,
-		MaxExpire:  maxExpire,
-		AuthHeader: p.principalHeader,
-	})
 	p.roleToken = roleToken
-
 	return nil
 }
 
@@ -164,8 +201,8 @@ func (p *athenzAuthProvider) Close() error {
 	return nil
 }
 
-func parseURI(uri string) privateKeyURI {
-	var uriSt privateKeyURI
+func parseURI(uri string) parsedURI {
+	var uriSt parsedURI
 	// scheme mediatype[;base64] path file
 	const expression = `^(?:([^:/?#]+):)(?:([;/\\\-\w]*),)?(?:/{0,2}((?:[^?#/]*/)*))?([^?#]*)`
 
@@ -177,17 +214,47 @@ func parseURI(uri string) privateKeyURI {
 		uriSt.MediaTypeAndEncodingType = groups[2]
 		uriSt.Data = groups[4]
 		uriSt.Path = groups[3] + groups[4]
+	} else {
+		// consider a file path specified instead of a URI
+		uriSt.Scheme = "file"
+		uriSt.Path = uri
 	}
 
 	return uriSt
 }
 
+func loadPEM(uri string) ([]byte, error) {
+	uriSt := parseURI(uri)
+	var pemData []byte
+
+	if uriSt.Scheme == "data" {
+		if uriSt.MediaTypeAndEncodingType != "application/x-pem-file;base64" {
+			return nil, errors.New("Unsupported mediaType or encodingType: " + uriSt.MediaTypeAndEncodingType)
+		}
+		pem, err := base64.StdEncoding.DecodeString(uriSt.Data)
+		if err != nil {
+			return nil, err
+		}
+		pemData = pem
+	} else if uriSt.Scheme == "file" {
+		pem, err := os.ReadFile(uriSt.Path)
+		if err != nil {
+			return nil, err
+		}
+		pemData = pem
+	} else {
+		return nil, errors.New("Unsupported URI Scheme: " + uriSt.Scheme)
+	}
+
+	return pemData, nil
+}
+
 func (p *athenzAuthProvider) RoundTrip(req *http.Request) (*http.Response, error) {
 	tok, err := p.roleToken.RoleTokenValue()
 	if err != nil {
 		return nil, err
 	}
-	req.Header.Add(AthenzRoleAuthHeader, tok)
+	req.Header.Add(p.roleHeader, tok)
 	return p.T.RoundTrip(req)
 }
 
diff --git a/pulsar/client.go b/pulsar/client.go
index d75c977..bc3f4f5 100644
--- a/pulsar/client.go
+++ b/pulsar/client.go
@@ -142,6 +142,16 @@ type ClientOptions struct {
 	// Specify metric registerer used to register metrics.
 	// Default prometheus.DefaultRegisterer
 	MetricsRegisterer prometheus.Registerer
+
+	// Release the connection if it is not used for more than ConnectionMaxIdleTime.
+	// Default is 60 seconds, negative such as -1 to disable.
+	ConnectionMaxIdleTime time.Duration
+
+	EnableTransaction bool
+
+	// Limit of client memory usage (in byte). The 64M default can guarantee a high producer throughput.
+	// Config less than 0 indicates off memory limit.
+	MemoryLimitBytes int64
 }
 
 // Client represents a pulsar client
diff --git a/pulsar/client_impl.go b/pulsar/client_impl.go
index e794e3d..7c8fcc9 100644
--- a/pulsar/client_impl.go
+++ b/pulsar/client_impl.go
@@ -33,6 +33,9 @@ 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
 )
 
 type client struct {
@@ -41,6 +44,8 @@ type client struct {
 	handlers      internal.ClientHandlers
 	lookupService internal.LookupService
 	metrics       *internal.Metrics
+	tcClient      *transactionCoordinatorClient
+	memLimit      internal.MemoryLimitController
 
 	log log.Logger
 }
@@ -53,6 +58,16 @@ func newClient(options ClientOptions) (Client, error) {
 		logger = log.NewLoggerWithLogrus(logrus.StandardLogger())
 	}
 
+	connectionMaxIdleTime := options.ConnectionMaxIdleTime
+	if connectionMaxIdleTime == 0 {
+		connectionMaxIdleTime = defaultConnMaxIdleTime
+	} else if connectionMaxIdleTime > 0 && connectionMaxIdleTime < minConnMaxIdleTime {
+		return nil, newError(InvalidConfiguration, fmt.Sprintf("Connection max idle time should be at least %f "+
+			"seconds", minConnMaxIdleTime.Seconds()))
+	} else {
+		logger.Debugf("Disable auto release idle connections")
+	}
+
 	if options.URL == "" {
 		return nil, newError(InvalidConfiguration, "URL is required for client")
 	}
@@ -133,11 +148,17 @@ func newClient(options ClientOptions) (Client, error) {
 		keepAliveInterval = defaultKeepAliveInterval
 	}
 
+	memLimitBytes := options.MemoryLimitBytes
+	if memLimitBytes == 0 {
+		memLimitBytes = defaultMemoryLimitBytes
+	}
+
 	c := &client{
 		cnxPool: internal.NewConnectionPool(tlsConfig, authProvider, connectionTimeout, keepAliveInterval,
-			maxConnectionsPerHost, logger, metrics),
-		log:     logger,
-		metrics: metrics,
+			maxConnectionsPerHost, logger, metrics, connectionMaxIdleTime),
+		log:      logger,
+		metrics:  metrics,
+		memLimit: internal.NewMemoryLimitController(memLimitBytes),
 	}
 	serviceNameResolver := internal.NewPulsarServiceNameResolver(url)
 
@@ -162,6 +183,14 @@ func newClient(options ClientOptions) (Client, error) {
 
 	c.handlers = internal.NewClientHandlers()
 
+	if options.EnableTransaction {
+		c.tcClient = newTransactionCoordinatorClientImpl(c)
+		err = c.tcClient.start()
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	return c, nil
 }
 
diff --git a/pulsar/consumer.go b/pulsar/consumer.go
index 8bae57d..9576d7a 100644
--- a/pulsar/consumer.go
+++ b/pulsar/consumer.go
@@ -80,6 +80,20 @@ type DLQPolicy struct {
 	RetryLetterTopic string
 }
 
+// AckGroupingOptions controls how to group ACK requests
+// If maxSize is 0 or 1, any ACK request will be sent immediately.
+// Otherwise, the ACK requests will be cached until one of the following conditions meets:
+// 1. There are `MaxSize` pending ACK requests.
+// 2. `MaxTime` is greater than 1 microsecond and ACK requests have been cached for `maxTime`.
+// Specially, for cumulative acknowledgment, only the latest ACK is cached and it will only be sent after `MaxTime`.
+type AckGroupingOptions struct {
+	// The maximum number of ACK requests to cache
+	MaxSize uint32
+
+	// The maximum time to cache ACK requests
+	MaxTime time.Duration
+}
+
 // ConsumerOptions is used to configure and create instances of Consumer.
 type ConsumerOptions struct {
 	// Topic specifies the topic this consumer will subscribe on.
@@ -215,6 +229,13 @@ type ConsumerOptions struct {
 	// Enable or disable batch index acknowledgment. To enable this feature, ensure batch index acknowledgment
 	// is enabled on the broker side. (default: false)
 	EnableBatchIndexAcknowledgment bool
+
+	// Controls how to group ACK requests, the default value is nil, which means:
+	// MaxSize: 1000
+	// MaxTime: 100*time.Millisecond
+	// NOTE: This option does not work if AckWithResponse is true
+	//	because there are only synchronous APIs for acknowledgment
+	AckGroupingOptions *AckGroupingOptions
 }
 
 // Consumer is an interface that abstracts behavior of Pulsar's consumer
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
index bf136c8..7a86574 100644
--- a/pulsar/consumer_impl.go
+++ b/pulsar/consumer_impl.go
@@ -383,7 +383,7 @@ func (c *consumer) internalTopicSubscribeToPartitions() error {
 				metadata:                    metadata,
 				subProperties:               subProperties,
 				replicateSubscriptionState:  c.options.ReplicateSubscriptionState,
-				startMessageID:              trackingMessageID{},
+				startMessageID:              nil,
 				subscriptionMode:            durable,
 				readCompacted:               c.options.ReadCompacted,
 				interceptors:                c.options.Interceptors,
@@ -398,6 +398,7 @@ func (c *consumer) internalTopicSubscribeToPartitions() error {
 				autoAckIncompleteChunk:      c.options.AutoAckIncompleteChunk,
 				consumerEventListener:       c.options.EventListener,
 				enableBatchIndexAck:         c.options.EnableBatchIndexAcknowledgment,
+				ackGroupingOptions:          c.options.AckGroupingOptions,
 			}
 			cons, err := newPartitionConsumer(c, c.client, opts, c.messageCh, c.dlq, c.metrics)
 			ch <- ConsumerError{
@@ -530,8 +531,8 @@ func (c *consumer) ReconsumeLaterWithCustomProperties(msg Message, customPropert
 	if delay < 0 {
 		delay = 0
 	}
-	msgID, ok := c.messageID(msg.ID())
-	if !ok {
+	msgID := c.messageID(msg.ID())
+	if msgID == nil {
 		return
 	}
 	props := make(map[string]string)
@@ -580,13 +581,13 @@ func (c *consumer) ReconsumeLaterWithCustomProperties(msg Message, customPropert
 
 func (c *consumer) Nack(msg Message) {
 	if c.options.EnableDefaultNackBackoffPolicy || c.options.NackBackoffPolicy != nil {
-		mid, ok := c.messageID(msg.ID())
-		if !ok {
+		mid := c.messageID(msg.ID())
+		if mid == nil {
 			return
 		}
 
 		if mid.consumer != nil {
-			mid.consumer.NackID(msg.ID())
+			mid.NackByMsg(msg)
 			return
 		}
 		c.consumers[mid.partitionIdx].NackMsg(msg)
@@ -742,11 +743,11 @@ func toProtoInitialPosition(p SubscriptionInitialPosition) pb.CommandSubscribe_I
 	return pb.CommandSubscribe_Latest
 }
 
-func (c *consumer) messageID(msgID MessageID) (trackingMessageID, bool) {
-	mid, ok := toTrackingMessageID(msgID)
-	if !ok {
+func (c *consumer) messageID(msgID MessageID) *trackingMessageID {
+	mid := toTrackingMessageID(msgID)
+	if mid == nil {
 		c.log.Warnf("invalid message id type %T", msgID)
-		return trackingMessageID{}, false
+		return nil
 	}
 
 	partition := int(mid.partitionIdx)
@@ -754,10 +755,10 @@ func (c *consumer) messageID(msgID MessageID) (trackingMessageID, bool) {
 	if partition < 0 || partition >= len(c.consumers) {
 		c.log.Warnf("invalid partition index %d expected a partition between [0-%d]",
 			partition, len(c.consumers))
-		return trackingMessageID{}, false
+		return nil
 	}
 
-	return mid, true
+	return mid
 }
 
 func addMessageCryptoIfMissing(client *client, options *ConsumerOptions, topics interface{}) error {
diff --git a/pulsar/consumer_multitopic.go b/pulsar/consumer_multitopic.go
index 452915a..c0fcaef 100644
--- a/pulsar/consumer_multitopic.go
+++ b/pulsar/consumer_multitopic.go
@@ -125,8 +125,8 @@ func (c *multiTopicConsumer) Ack(msg Message) error {
 
 // AckID the consumption of a single message, identified by its MessageID
 func (c *multiTopicConsumer) AckID(msgID MessageID) error {
-	mid, ok := toTrackingMessageID(msgID)
-	if !ok {
+	mid := toTrackingMessageID(msgID)
+	if mid == nil {
 		c.log.Warnf("invalid message id type %T", msgID)
 		return errors.New("invalid message id type in multi_consumer")
 	}
@@ -152,8 +152,8 @@ func (c *multiTopicConsumer) AckCumulative(msg Message) error {
 // AckIDCumulative the reception of all the messages in the stream up to (and including)
 // the provided message, identified by its MessageID
 func (c *multiTopicConsumer) AckIDCumulative(msgID MessageID) error {
-	mid, ok := toTrackingMessageID(msgID)
-	if !ok {
+	mid := toTrackingMessageID(msgID)
+	if mid == nil {
 		c.log.Warnf("invalid message id type %T", msgID)
 		return errors.New("invalid message id type in multi_consumer")
 	}
@@ -203,8 +203,8 @@ func (c *multiTopicConsumer) ReconsumeLaterWithCustomProperties(msg Message, cus
 func (c *multiTopicConsumer) Nack(msg Message) {
 	if c.options.EnableDefaultNackBackoffPolicy || c.options.NackBackoffPolicy != nil {
 		msgID := msg.ID()
-		mid, ok := toTrackingMessageID(msgID)
-		if !ok {
+		mid := toTrackingMessageID(msgID)
+		if mid == nil {
 			c.log.Warnf("invalid message id type %T", msgID)
 			return
 		}
@@ -221,8 +221,8 @@ func (c *multiTopicConsumer) Nack(msg Message) {
 }
 
 func (c *multiTopicConsumer) NackID(msgID MessageID) {
-	mid, ok := toTrackingMessageID(msgID)
-	if !ok {
+	mid := toTrackingMessageID(msgID)
+	if mid == nil {
 		c.log.Warnf("invalid message id type %T", msgID)
 		return
 	}
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index e723f8a..0f7af3d 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -41,10 +41,6 @@ import (
 	uAtomic "go.uber.org/atomic"
 )
 
-var (
-	lastestMessageID = LatestMessageID()
-)
-
 type consumerState int
 
 const (
@@ -98,7 +94,7 @@ type partitionConsumerOpts struct {
 	metadata                    map[string]string
 	subProperties               map[string]string
 	replicateSubscriptionState  bool
-	startMessageID              trackingMessageID
+	startMessageID              *trackingMessageID
 	startMessageIDInclusive     bool
 	subscriptionMode            subscriptionMode
 	readCompacted               bool
@@ -116,6 +112,7 @@ type partitionConsumerOpts struct {
 	// in failover mode, this callback will be called when consumer change
 	consumerEventListener ConsumerEventListener
 	enableBatchIndexAck   bool
+	ackGroupingOptions    *AckGroupingOptions
 }
 
 type ConsumerEventListener interface {
@@ -148,13 +145,13 @@ type partitionConsumer struct {
 	queueSize       int32
 	queueCh         chan []*message
 	startMessageID  atomicMessageID
-	lastDequeuedMsg trackingMessageID
+	lastDequeuedMsg *trackingMessageID
 
 	eventsCh        chan interface{}
 	connectedCh     chan struct{}
 	connectClosedCh chan connectionClosed
 	closeCh         chan struct{}
-	clearQueueCh    chan func(id trackingMessageID)
+	clearQueueCh    chan func(id *trackingMessageID)
 
 	nackTracker *negativeAcksTracker
 	dlq         *dlqRouter
@@ -167,6 +164,7 @@ type partitionConsumer struct {
 
 	chunkedMsgCtxMap   *chunkedMsgCtxMap
 	unAckChunksTracker *unAckChunksTracker
+	ackGroupingTracker ackGroupingTracker
 }
 
 func (pc *partitionConsumer) ActiveConsumerChanged(isActive bool) {
@@ -215,17 +213,17 @@ func (p *availablePermits) reset() {
 
 // atomicMessageID is a wrapper for trackingMessageID to make get and set atomic
 type atomicMessageID struct {
-	msgID trackingMessageID
+	msgID *trackingMessageID
 	sync.RWMutex
 }
 
-func (a *atomicMessageID) get() trackingMessageID {
+func (a *atomicMessageID) get() *trackingMessageID {
 	a.RLock()
 	defer a.RUnlock()
 	return a.msgID
 }
 
-func (a *atomicMessageID) set(msgID trackingMessageID) {
+func (a *atomicMessageID) set(msgID *trackingMessageID) {
 	a.Lock()
 	defer a.Unlock()
 	a.msgID = msgID
@@ -301,7 +299,7 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 		messageCh:            messageCh,
 		connectClosedCh:      make(chan connectionClosed, 10),
 		closeCh:              make(chan struct{}),
-		clearQueueCh:         make(chan func(id trackingMessageID)),
+		clearQueueCh:         make(chan func(id *trackingMessageID)),
 		compressionProviders: sync.Map{},
 		dlq:                  dlq,
 		metrics:              metrics,
@@ -310,6 +308,9 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 	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) })
 	pc.setConsumerState(consumerInit)
 	pc.log = client.log.SubLogger(log.Fields{
 		"name":         pc.name,
@@ -342,7 +343,8 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 	pc.log.Info("Created consumer")
 	pc.setConsumerState(consumerReady)
 
-	if pc.options.startMessageIDInclusive && pc.startMessageID.get().equal(lastestMessageID.(messageID)) {
+	startingMessageID := pc.startMessageID.get()
+	if pc.options.startMessageIDInclusive && startingMessageID != nil && startingMessageID.equal(latestMessageID) {
 		msgID, err := pc.requestGetLastMessageID()
 		if err != nil {
 			pc.nackTracker.Close()
@@ -413,10 +415,10 @@ func (pc *partitionConsumer) internalUnsubscribe(unsub *unsubscribeRequest) {
 	pc.setConsumerState(consumerClosed)
 }
 
-func (pc *partitionConsumer) getLastMessageID() (trackingMessageID, error) {
+func (pc *partitionConsumer) getLastMessageID() (*trackingMessageID, error) {
 	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
 		pc.log.WithField("state", state).Error("Failed to redeliver closing or closed consumer")
-		return trackingMessageID{}, errors.New("failed to redeliver closing or closed consumer")
+		return nil, errors.New("failed to redeliver closing or closed consumer")
 	}
 	req := &getLastMsgIDRequest{doneCh: make(chan struct{})}
 	pc.eventsCh <- req
@@ -431,10 +433,10 @@ func (pc *partitionConsumer) internalGetLastMessageID(req *getLastMsgIDRequest)
 	req.msgID, req.err = pc.requestGetLastMessageID()
 }
 
-func (pc *partitionConsumer) requestGetLastMessageID() (trackingMessageID, error) {
+func (pc *partitionConsumer) requestGetLastMessageID() (*trackingMessageID, error) {
 	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
 		pc.log.WithField("state", state).Error("Failed to getLastMessageID closing or closed consumer")
-		return trackingMessageID{}, errors.New("failed to getLastMessageID closing or closed consumer")
+		return nil, errors.New("failed to getLastMessageID closing or closed consumer")
 	}
 
 	requestID := pc.client.rpcClient.NewRequestID()
@@ -446,7 +448,7 @@ func (pc *partitionConsumer) requestGetLastMessageID() (trackingMessageID, error
 		pb.BaseCommand_GET_LAST_MESSAGE_ID, cmdGetLastMessageID)
 	if err != nil {
 		pc.log.WithError(err).Error("Failed to get last message id")
-		return trackingMessageID{}, err
+		return nil, err
 	}
 	id := res.Response.GetLastMessageIdResponse.GetLastMessageId()
 	return convertToMessageID(id), nil
@@ -458,39 +460,44 @@ func (pc *partitionConsumer) ackID(msgID MessageID, withResponse bool) error {
 		return errors.New("consumer state is closed")
 	}
 
-	if cmid, ok := toChunkedMessageID(msgID); ok {
+	if cmid, ok := msgID.(*chunkMessageID); ok {
 		return pc.unAckChunksTracker.ack(cmid)
 	}
 
-	trackingID, ok := toTrackingMessageID(msgID)
-	if !ok {
+	trackingID := toTrackingMessageID(msgID)
+	if trackingID == nil {
 		return errors.New("failed to convert trackingMessageID")
 	}
 
-	ackReq := new(ackRequest)
-	ackReq.doneCh = make(chan struct{})
-	ackReq.ackType = individualAck
-	if !trackingID.Undefined() && trackingID.ack() {
+	if trackingID != nil && trackingID.ack() {
 		pc.metrics.AcksCounter.Inc()
 		pc.metrics.ProcessingTime.Observe(float64(time.Now().UnixNano()-trackingID.receivedTime.UnixNano()) / 1.0e9)
-		ackReq.msgID = trackingID
-		// send ack request to eventsCh
-		pc.eventsCh <- ackReq
-
-		if withResponse {
-			<-ackReq.doneCh
-		}
-
-		pc.options.interceptors.OnAcknowledge(pc.parentConsumer, msgID)
-	} else if pc.options.enableBatchIndexAck {
-		ackReq.msgID = trackingID
-		pc.eventsCh <- ackReq
+	} else if !pc.options.enableBatchIndexAck {
+		return nil
 	}
 
+	var ackReq *ackRequest
 	if withResponse {
-		return ackReq.err
+		ackReq := pc.sendIndividualAck(trackingID)
+		<-ackReq.doneCh
+	} else {
+		pc.ackGroupingTracker.add(trackingID)
 	}
-	return nil
+	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{}),
+		ackType: individualAck,
+		msgID:   *msgID.(*trackingMessageID),
+	}
+	pc.eventsCh <- ackReq
+	return ackReq
 }
 
 func (pc *partitionConsumer) AckIDWithResponse(msgID MessageID) error {
@@ -516,22 +523,17 @@ func (pc *partitionConsumer) internalAckIDCumulative(msgID MessageID, withRespon
 	}
 
 	// chunk message id will be converted to tracking message id
-	trackingID, ok := toTrackingMessageID(msgID)
-	if !ok {
+	trackingID := toTrackingMessageID(msgID)
+	if trackingID == nil {
 		return errors.New("failed to convert trackingMessageID")
 	}
-	if trackingID.Undefined() {
-		return nil
-	}
 
-	ackReq := new(ackRequest)
-	ackReq.doneCh = make(chan struct{})
-	ackReq.ackType = cumulativeAck
+	var msgIDToAck *trackingMessageID
 	if trackingID.ackCumulative() || pc.options.enableBatchIndexAck {
-		ackReq.msgID = trackingID
+		msgIDToAck = trackingID
 	} else if !trackingID.tracker.hasPrevBatchAcked() {
 		// get previous batch message id
-		ackReq.msgID = trackingID.prev()
+		msgIDToAck = trackingID.prev()
 		trackingID.tracker.setPrevBatchAcked()
 	} else {
 		// waiting for all the msgs are acked in this batch
@@ -540,31 +542,45 @@ func (pc *partitionConsumer) internalAckIDCumulative(msgID MessageID, withRespon
 
 	pc.metrics.AcksCounter.Inc()
 	pc.metrics.ProcessingTime.Observe(float64(time.Now().UnixNano()-trackingID.receivedTime.UnixNano()) / 1.0e9)
-	// send ack request to eventsCh
-	pc.eventsCh <- ackReq
 
+	var ackReq *ackRequest
 	if withResponse {
-		// wait for the request to complete if withResponse set true
+		ackReq := pc.sendCumulativeAck(msgIDToAck)
 		<-ackReq.doneCh
+	} else {
+		pc.ackGroupingTracker.addCumulative(msgIDToAck)
 	}
 
 	pc.options.interceptors.OnAcknowledge(pc.parentConsumer, msgID)
 
-	if cmid, ok := toChunkedMessageID(msgID); ok {
+	if cmid, ok := msgID.(*chunkMessageID); ok {
 		pc.unAckChunksTracker.remove(cmid)
 	}
 
-	return nil
+	if ackReq == nil {
+		return nil
+	}
+	return ackReq.err
+}
+
+func (pc *partitionConsumer) sendCumulativeAck(msgID MessageID) *ackRequest {
+	ackReq := &ackRequest{
+		doneCh:  make(chan struct{}),
+		ackType: cumulativeAck,
+		msgID:   *msgID.(*trackingMessageID),
+	}
+	pc.eventsCh <- ackReq
+	return ackReq
 }
 
 func (pc *partitionConsumer) NackID(msgID MessageID) {
-	if cmid, ok := toChunkedMessageID(msgID); ok {
+	if cmid, ok := msgID.(*chunkMessageID); ok {
 		pc.unAckChunksTracker.nack(cmid)
 		return
 	}
 
-	trackingID, ok := toTrackingMessageID(msgID)
-	if !ok {
+	trackingID := toTrackingMessageID(msgID)
+	if trackingID == nil {
 		return
 	}
 
@@ -631,6 +647,9 @@ func (pc *partitionConsumer) Close() {
 		return
 	}
 
+	// flush all pending ACK requests and terminate the timer goroutine
+	pc.ackGroupingTracker.close()
+
 	// close chunkedMsgCtxMap
 	pc.chunkedMsgCtxMap.Close()
 
@@ -649,15 +668,16 @@ func (pc *partitionConsumer) Seek(msgID MessageID) error {
 	req := &seekRequest{
 		doneCh: make(chan struct{}),
 	}
-	if cmid, ok := toChunkedMessageID(msgID); ok {
+	if cmid, ok := msgID.(*chunkMessageID); ok {
 		req.msgID = cmid.firstChunkID
-	} else if tmid, ok := toTrackingMessageID(msgID); ok {
+	} else if tmid := toTrackingMessageID(msgID); tmid != nil {
 		req.msgID = tmid.messageID
 	} else {
 		// will never reach
 		return errors.New("unhandled messageID type")
 	}
 
+	pc.ackGroupingTracker.flushAndClean()
 	pc.eventsCh <- req
 
 	// wait for the request to complete
@@ -669,7 +689,7 @@ func (pc *partitionConsumer) internalSeek(seek *seekRequest) {
 	defer close(seek.doneCh)
 	seek.err = pc.requestSeek(seek.msgID)
 }
-func (pc *partitionConsumer) requestSeek(msgID messageID) error {
+func (pc *partitionConsumer) requestSeek(msgID *messageID) error {
 	if err := pc.requestSeekWithoutClear(msgID); err != nil {
 		return err
 	}
@@ -677,7 +697,7 @@ func (pc *partitionConsumer) requestSeek(msgID messageID) error {
 	return nil
 }
 
-func (pc *partitionConsumer) requestSeekWithoutClear(msgID messageID) error {
+func (pc *partitionConsumer) requestSeekWithoutClear(msgID *messageID) error {
 	state := pc.getConsumerState()
 	if state == consumerClosing || state == consumerClosed {
 		pc.log.WithField("state", state).Error("failed seek by consumer is closing or has closed")
@@ -715,6 +735,7 @@ func (pc *partitionConsumer) SeekByTime(time time.Time) error {
 		doneCh:      make(chan struct{}),
 		publishTime: time,
 	}
+	pc.ackGroupingTracker.flushAndClean()
 	pc.eventsCh <- req
 
 	// wait for the request to complete
@@ -957,6 +978,10 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 			msgID = trackingMsgID
 		}
 
+		if pc.ackGroupingTracker.isDuplicate(msgID) {
+			continue
+		}
+
 		var messageIndex *uint64
 		var brokerPublishTime *time.Time
 		if brokerMetadata != nil {
@@ -1032,7 +1057,7 @@ func (pc *partitionConsumer) processMessageChunk(compressedPayload internal.Buff
 	numChunks := msgMeta.GetNumChunksFromMsg()
 	totalChunksSize := int(msgMeta.GetTotalChunkMsgSize())
 	chunkID := msgMeta.GetChunkId()
-	msgID := messageID{
+	msgID := &messageID{
 		ledgerID:     int64(pbMsgID.GetLedgerId()),
 		entryID:      int64(pbMsgID.GetEntryId()),
 		batchIdx:     -1,
@@ -1074,12 +1099,12 @@ func (pc *partitionConsumer) processMessageChunk(compressedPayload internal.Buff
 	return ctx.chunkedMsgBuffer
 }
 
-func (pc *partitionConsumer) messageShouldBeDiscarded(msgID trackingMessageID) bool {
-	if pc.startMessageID.get().Undefined() {
+func (pc *partitionConsumer) messageShouldBeDiscarded(msgID *trackingMessageID) bool {
+	if pc.startMessageID.get() == nil {
 		return false
 	}
 	// if we start at latest message, we should never discard
-	if pc.options.startMessageID.equal(latestMessageID) {
+	if pc.options.startMessageID != nil && pc.options.startMessageID.equal(latestMessageID) {
 		return false
 	}
 
@@ -1232,7 +1257,7 @@ func (pc *partitionConsumer) dispatcher() {
 		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
-			var nextMessageInQueue trackingMessageID
+			var nextMessageInQueue *trackingMessageID
 			go func() {
 				pc.queueCh <- nil
 			}()
@@ -1241,8 +1266,8 @@ func (pc *partitionConsumer) dispatcher() {
 				// the queue has been drained
 				if m == nil {
 					break
-				} else if nextMessageInQueue.Undefined() {
-					nextMessageInQueue, _ = toTrackingMessageID(m[0].msgID)
+				} else if nextMessageInQueue == nil {
+					nextMessageInQueue = toTrackingMessageID(m[0].msgID)
 				}
 			}
 
@@ -1280,13 +1305,13 @@ type redeliveryRequest struct {
 
 type getLastMsgIDRequest struct {
 	doneCh chan struct{}
-	msgID  trackingMessageID
+	msgID  *trackingMessageID
 	err    error
 }
 
 type seekRequest struct {
 	doneCh chan struct{}
-	msgID  messageID
+	msgID  *messageID
 	err    error
 }
 
@@ -1547,15 +1572,15 @@ func (pc *partitionConsumer) grabConn() error {
 	}
 }
 
-func (pc *partitionConsumer) clearQueueAndGetNextMessage() trackingMessageID {
+func (pc *partitionConsumer) clearQueueAndGetNextMessage() *trackingMessageID {
 	if pc.getConsumerState() != consumerReady {
-		return trackingMessageID{}
+		return nil
 	}
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
-	var msgID trackingMessageID
+	var msgID *trackingMessageID
 
-	pc.clearQueueCh <- func(id trackingMessageID) {
+	pc.clearQueueCh <- func(id *trackingMessageID) {
 		msgID = id
 		wg.Done()
 	}
@@ -1568,16 +1593,16 @@ func (pc *partitionConsumer) clearQueueAndGetNextMessage() trackingMessageID {
  * Clear the internal receiver queue and returns the message id of what was the 1st message in the queue that was
  * not seen by the application
  */
-func (pc *partitionConsumer) clearReceiverQueue() trackingMessageID {
+func (pc *partitionConsumer) clearReceiverQueue() *trackingMessageID {
 	nextMessageInQueue := pc.clearQueueAndGetNextMessage()
 
-	if pc.startMessageID.get().Undefined() {
+	if pc.startMessageID.get() == nil {
 		return pc.startMessageID.get()
 	}
 
-	if !nextMessageInQueue.Undefined() {
+	if nextMessageInQueue != nil {
 		return getPreviousMessage(nextMessageInQueue)
-	} else if !pc.lastDequeuedMsg.Undefined() {
+	} else if pc.lastDequeuedMsg != nil {
 		// If the queue was empty we need to restart from the message just after the last one that has been dequeued
 		// in the past
 		return pc.lastDequeuedMsg
@@ -1587,10 +1612,10 @@ func (pc *partitionConsumer) clearReceiverQueue() trackingMessageID {
 	}
 }
 
-func getPreviousMessage(mid trackingMessageID) trackingMessageID {
+func getPreviousMessage(mid *trackingMessageID) *trackingMessageID {
 	if mid.batchIdx >= 0 {
-		return trackingMessageID{
-			messageID: messageID{
+		return &trackingMessageID{
+			messageID: &messageID{
 				ledgerID:     mid.ledgerID,
 				entryID:      mid.entryID,
 				batchIdx:     mid.batchIdx - 1,
@@ -1603,8 +1628,8 @@ func getPreviousMessage(mid trackingMessageID) trackingMessageID {
 	}
 
 	// Get on previous message in previous entry
-	return trackingMessageID{
-		messageID: messageID{
+	return &trackingMessageID{
+		messageID: &messageID{
 			ledgerID:     mid.ledgerID,
 			entryID:      mid.entryID - 1,
 			batchIdx:     mid.batchIdx,
@@ -1703,8 +1728,8 @@ func (pc *partitionConsumer) _getConn() internal.Connection {
 	return pc.conn.Load().(internal.Connection)
 }
 
-func convertToMessageIDData(msgID trackingMessageID) *pb.MessageIdData {
-	if msgID.Undefined() {
+func convertToMessageIDData(msgID *trackingMessageID) *pb.MessageIdData {
+	if msgID == nil {
 		return nil
 	}
 
@@ -1714,13 +1739,13 @@ func convertToMessageIDData(msgID trackingMessageID) *pb.MessageIdData {
 	}
 }
 
-func convertToMessageID(id *pb.MessageIdData) trackingMessageID {
+func convertToMessageID(id *pb.MessageIdData) *trackingMessageID {
 	if id == nil {
-		return trackingMessageID{}
+		return nil
 	}
 
-	msgID := trackingMessageID{
-		messageID: messageID{
+	msgID := &trackingMessageID{
+		messageID: &messageID{
 			ledgerID: int64(*id.LedgerId),
 			entryID:  int64(*id.EntryId),
 		},
@@ -1736,7 +1761,7 @@ type chunkedMsgCtx struct {
 	totalChunks      int32
 	chunkedMsgBuffer internal.Buffer
 	lastChunkedMsgID int32
-	chunkedMsgIDs    []messageID
+	chunkedMsgIDs    []*messageID
 	receivedTime     int64
 
 	mu sync.Mutex
@@ -1747,12 +1772,12 @@ func newChunkedMsgCtx(numChunksFromMsg int32, totalChunkMsgSize int) *chunkedMsg
 		totalChunks:      numChunksFromMsg,
 		chunkedMsgBuffer: internal.NewBuffer(totalChunkMsgSize),
 		lastChunkedMsgID: -1,
-		chunkedMsgIDs:    make([]messageID, numChunksFromMsg),
+		chunkedMsgIDs:    make([]*messageID, numChunksFromMsg),
 		receivedTime:     time.Now().Unix(),
 	}
 }
 
-func (c *chunkedMsgCtx) append(chunkID int32, msgID messageID, partPayload internal.Buffer) {
+func (c *chunkedMsgCtx) append(chunkID int32, msgID *messageID, partPayload internal.Buffer) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	c.chunkedMsgIDs[chunkID] = msgID
@@ -1760,20 +1785,20 @@ func (c *chunkedMsgCtx) append(chunkID int32, msgID messageID, partPayload inter
 	c.lastChunkedMsgID = chunkID
 }
 
-func (c *chunkedMsgCtx) firstChunkID() messageID {
+func (c *chunkedMsgCtx) firstChunkID() *messageID {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	if len(c.chunkedMsgIDs) == 0 {
-		return messageID{}
+		return nil
 	}
 	return c.chunkedMsgIDs[0]
 }
 
-func (c *chunkedMsgCtx) lastChunkID() messageID {
+func (c *chunkedMsgCtx) lastChunkID() *messageID {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	if len(c.chunkedMsgIDs) == 0 {
-		return messageID{}
+		return nil
 	}
 	return c.chunkedMsgIDs[len(c.chunkedMsgIDs)-1]
 }
@@ -1783,9 +1808,13 @@ func (c *chunkedMsgCtx) discard(pc *partitionConsumer) {
 	defer c.mu.Unlock()
 
 	for _, mid := range c.chunkedMsgIDs {
+		if mid == nil {
+			continue
+		}
 		pc.log.Info("Removing chunk message-id", mid.String())
-		tmid, _ := toTrackingMessageID(mid)
-		pc.AckID(tmid)
+		if tmid := toTrackingMessageID(mid); tmid != nil {
+			pc.AckID(tmid)
+		}
 	}
 }
 
@@ -1904,40 +1933,41 @@ func (c *chunkedMsgCtxMap) Close() {
 }
 
 type unAckChunksTracker struct {
-	chunkIDs map[chunkMessageID][]messageID
+	// TODO: use hash code of chunkMessageID as the key
+	chunkIDs map[chunkMessageID][]*messageID
 	pc       *partitionConsumer
 	mu       sync.Mutex
 }
 
 func newUnAckChunksTracker(pc *partitionConsumer) *unAckChunksTracker {
 	return &unAckChunksTracker{
-		chunkIDs: make(map[chunkMessageID][]messageID),
+		chunkIDs: make(map[chunkMessageID][]*messageID),
 		pc:       pc,
 	}
 }
 
-func (u *unAckChunksTracker) add(cmid chunkMessageID, ids []messageID) {
+func (u *unAckChunksTracker) add(cmid *chunkMessageID, ids []*messageID) {
 	u.mu.Lock()
 	defer u.mu.Unlock()
 
-	u.chunkIDs[cmid] = ids
+	u.chunkIDs[*cmid] = ids
 }
 
-func (u *unAckChunksTracker) get(cmid chunkMessageID) []messageID {
+func (u *unAckChunksTracker) get(cmid *chunkMessageID) []*messageID {
 	u.mu.Lock()
 	defer u.mu.Unlock()
 
-	return u.chunkIDs[cmid]
+	return u.chunkIDs[*cmid]
 }
 
-func (u *unAckChunksTracker) remove(cmid chunkMessageID) {
+func (u *unAckChunksTracker) remove(cmid *chunkMessageID) {
 	u.mu.Lock()
 	defer u.mu.Unlock()
 
-	delete(u.chunkIDs, cmid)
+	delete(u.chunkIDs, *cmid)
 }
 
-func (u *unAckChunksTracker) ack(cmid chunkMessageID) error {
+func (u *unAckChunksTracker) ack(cmid *chunkMessageID) error {
 	ids := u.get(cmid)
 	for _, id := range ids {
 		if err := u.pc.AckID(id); err != nil {
@@ -1948,7 +1978,7 @@ func (u *unAckChunksTracker) ack(cmid chunkMessageID) error {
 	return nil
 }
 
-func (u *unAckChunksTracker) nack(cmid chunkMessageID) {
+func (u *unAckChunksTracker) nack(cmid *chunkMessageID) {
 	ids := u.get(cmid)
 	for _, id := range ids {
 		u.pc.NackID(id)
diff --git a/pulsar/consumer_regex.go b/pulsar/consumer_regex.go
index d890c67..fdfecec 100644
--- a/pulsar/consumer_regex.go
+++ b/pulsar/consumer_regex.go
@@ -174,8 +174,8 @@ func (c *regexConsumer) ReconsumeLaterWithCustomProperties(msg Message, customPr
 
 // AckID the consumption of a single message, identified by its MessageID
 func (c *regexConsumer) AckID(msgID MessageID) error {
-	mid, ok := toTrackingMessageID(msgID)
-	if !ok {
+	mid := toTrackingMessageID(msgID)
+	if mid == nil {
 		c.log.Warnf("invalid message id type %T", msgID)
 		return errors.New("invalid message id type")
 	}
@@ -201,8 +201,8 @@ func (c *regexConsumer) AckCumulative(msg Message) error {
 // AckIDCumulative the reception of all the messages in the stream up to (and including)
 // the provided message, identified by its MessageID
 func (c *regexConsumer) AckIDCumulative(msgID MessageID) error {
-	mid, ok := toTrackingMessageID(msgID)
-	if !ok {
+	mid := toTrackingMessageID(msgID)
+	if mid == nil {
 		c.log.Warnf("invalid message id type %T", msgID)
 		return errors.New("invalid message id type")
 	}
@@ -222,8 +222,8 @@ func (c *regexConsumer) AckIDCumulative(msgID MessageID) error {
 func (c *regexConsumer) Nack(msg Message) {
 	if c.options.EnableDefaultNackBackoffPolicy || c.options.NackBackoffPolicy != nil {
 		msgID := msg.ID()
-		mid, ok := toTrackingMessageID(msgID)
-		if !ok {
+		mid := toTrackingMessageID(msgID)
+		if mid == nil {
 			c.log.Warnf("invalid message id type %T", msgID)
 			return
 		}
@@ -240,8 +240,8 @@ func (c *regexConsumer) Nack(msg Message) {
 }
 
 func (c *regexConsumer) NackID(msgID MessageID) {
-	mid, ok := toTrackingMessageID(msgID)
-	if !ok {
+	mid := toTrackingMessageID(msgID)
+	if mid == nil {
 		c.log.Warnf("invalid message id type %T", msgID)
 		return
 	}
diff --git a/pulsar/error.go b/pulsar/error.go
index ead5cf9..0aa1e3c 100644
--- a/pulsar/error.go
+++ b/pulsar/error.go
@@ -103,6 +103,16 @@ 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
+
+	// ClientMemoryBufferIsFull client limit buffer is full
+	ClientMemoryBufferIsFull
 )
 
 // Error implement error interface, composed of two parts: msg and result.
@@ -209,6 +219,8 @@ func getResultStr(r Result) string {
 		return "ProducerClosed"
 	case SchemaFailure:
 		return "SchemaFailure"
+	case ClientMemoryBufferIsFull:
+		return "ClientMemoryBufferIsFull"
 	default:
 		return fmt.Sprintf("Result(%d)", r)
 	}
diff --git a/pulsar/impl_message.go b/pulsar/impl_message.go
index 39db8e1..68ddecd 100644
--- a/pulsar/impl_message.go
+++ b/pulsar/impl_message.go
@@ -39,7 +39,7 @@ type messageID struct {
 	batchSize    int32
 }
 
-var latestMessageID = messageID{
+var latestMessageID = &messageID{
 	ledgerID:     math.MaxInt64,
 	entryID:      math.MaxInt64,
 	batchIdx:     -1,
@@ -47,7 +47,7 @@ var latestMessageID = messageID{
 	batchSize:    0,
 }
 
-var earliestMessageID = messageID{
+var earliestMessageID = &messageID{
 	ledgerID:     -1,
 	entryID:      -1,
 	batchIdx:     -1,
@@ -56,18 +56,14 @@ var earliestMessageID = messageID{
 }
 
 type trackingMessageID struct {
-	messageID
+	*messageID
 
 	tracker      *ackTracker
 	consumer     acker
 	receivedTime time.Time
 }
 
-func (id trackingMessageID) Undefined() bool {
-	return id == trackingMessageID{}
-}
-
-func (id trackingMessageID) Ack() error {
+func (id *trackingMessageID) Ack() error {
 	if id.consumer == nil {
 		return errors.New("consumer is nil in trackingMessageID")
 	}
@@ -78,7 +74,7 @@ func (id trackingMessageID) Ack() error {
 	return nil
 }
 
-func (id trackingMessageID) AckWithResponse() error {
+func (id *trackingMessageID) AckWithResponse() error {
 	if id.consumer == nil {
 		return errors.New("consumer is nil in trackingMessageID")
 	}
@@ -89,37 +85,37 @@ func (id trackingMessageID) AckWithResponse() error {
 	return nil
 }
 
-func (id trackingMessageID) Nack() {
+func (id *trackingMessageID) Nack() {
 	if id.consumer == nil {
 		return
 	}
 	id.consumer.NackID(id)
 }
 
-func (id trackingMessageID) NackByMsg(msg Message) {
+func (id *trackingMessageID) NackByMsg(msg Message) {
 	if id.consumer == nil {
 		return
 	}
 	id.consumer.NackMsg(msg)
 }
 
-func (id trackingMessageID) ack() bool {
+func (id *trackingMessageID) ack() bool {
 	if id.tracker != nil && id.batchIdx > -1 {
 		return id.tracker.ack(int(id.batchIdx))
 	}
 	return true
 }
 
-func (id trackingMessageID) ackCumulative() bool {
+func (id *trackingMessageID) ackCumulative() bool {
 	if id.tracker != nil && id.batchIdx > -1 {
 		return id.tracker.ackCumulative(int(id.batchIdx))
 	}
 	return true
 }
 
-func (id trackingMessageID) prev() trackingMessageID {
-	return trackingMessageID{
-		messageID: messageID{
+func (id *trackingMessageID) prev() *trackingMessageID {
+	return &trackingMessageID{
+		messageID: &messageID{
 			ledgerID:     id.ledgerID,
 			entryID:      id.entryID - 1,
 			partitionIdx: id.partitionIdx,
@@ -129,11 +125,11 @@ func (id trackingMessageID) prev() trackingMessageID {
 	}
 }
 
-func (id messageID) isEntryIDValid() bool {
+func (id *messageID) isEntryIDValid() bool {
 	return id.entryID >= 0
 }
 
-func (id messageID) greater(other messageID) bool {
+func (id *messageID) greater(other *messageID) bool {
 	if id.ledgerID != other.ledgerID {
 		return id.ledgerID > other.ledgerID
 	}
@@ -145,17 +141,17 @@ func (id messageID) greater(other messageID) bool {
 	return id.batchIdx > other.batchIdx
 }
 
-func (id messageID) equal(other messageID) bool {
+func (id *messageID) equal(other *messageID) bool {
 	return id.ledgerID == other.ledgerID &&
 		id.entryID == other.entryID &&
 		id.batchIdx == other.batchIdx
 }
 
-func (id messageID) greaterEqual(other messageID) bool {
+func (id *messageID) greaterEqual(other *messageID) bool {
 	return id.equal(other) || id.greater(other)
 }
 
-func (id messageID) Serialize() []byte {
+func (id *messageID) Serialize() []byte {
 	msgID := &pb.MessageIdData{
 		LedgerId:   proto.Uint64(uint64(id.ledgerID)),
 		EntryId:    proto.Uint64(uint64(id.entryID)),
@@ -167,27 +163,27 @@ func (id messageID) Serialize() []byte {
 	return data
 }
 
-func (id messageID) LedgerID() int64 {
+func (id *messageID) LedgerID() int64 {
 	return id.ledgerID
 }
 
-func (id messageID) EntryID() int64 {
+func (id *messageID) EntryID() int64 {
 	return id.entryID
 }
 
-func (id messageID) BatchIdx() int32 {
+func (id *messageID) BatchIdx() int32 {
 	return id.batchIdx
 }
 
-func (id messageID) PartitionIdx() int32 {
+func (id *messageID) PartitionIdx() int32 {
 	return id.partitionIdx
 }
 
-func (id messageID) BatchSize() int32 {
+func (id *messageID) BatchSize() int32 {
 	return id.batchSize
 }
 
-func (id messageID) String() string {
+func (id *messageID) String() string {
 	return fmt.Sprintf("%d:%d:%d", id.ledgerID, id.entryID, id.partitionIdx)
 }
 
@@ -208,7 +204,7 @@ func deserializeMessageID(data []byte) (MessageID, error) {
 }
 
 func newMessageID(ledgerID int64, entryID int64, batchIdx int32, partitionIdx int32, batchSize int32) MessageID {
-	return messageID{
+	return &messageID{
 		ledgerID:     ledgerID,
 		entryID:      entryID,
 		batchIdx:     batchIdx,
@@ -218,9 +214,9 @@ func newMessageID(ledgerID int64, entryID int64, batchIdx int32, partitionIdx in
 }
 
 func newTrackingMessageID(ledgerID int64, entryID int64, batchIdx int32, partitionIdx int32, batchSize int32,
-	tracker *ackTracker) trackingMessageID {
-	return trackingMessageID{
-		messageID: messageID{
+	tracker *ackTracker) *trackingMessageID {
+	return &trackingMessageID{
+		messageID: &messageID{
 			ledgerID:     ledgerID,
 			entryID:      entryID,
 			batchIdx:     batchIdx,
@@ -232,31 +228,23 @@ func newTrackingMessageID(ledgerID int64, entryID int64, batchIdx int32, partiti
 	}
 }
 
-func toTrackingMessageID(msgID MessageID) (trackingMessageID, bool) {
-	if mid, ok := msgID.(messageID); ok {
-		return trackingMessageID{
+func toTrackingMessageID(msgID MessageID) *trackingMessageID {
+	if mid, ok := msgID.(*messageID); ok {
+		return &trackingMessageID{
 			messageID:    mid,
 			receivedTime: time.Now(),
-		}, true
-	} else if mid, ok := msgID.(trackingMessageID); ok {
-		return mid, true
-	} else if cmid, ok := msgID.(chunkMessageID); ok {
-		return trackingMessageID{
+		}
+	} else if mid, ok := msgID.(*trackingMessageID); ok {
+		return mid
+	} else if cmid, ok := msgID.(*chunkMessageID); ok {
+		return &trackingMessageID{
 			messageID:    cmid.messageID,
 			receivedTime: cmid.receivedTime,
 			consumer:     cmid.consumer,
-		}, true
+		}
 	} else {
-		return trackingMessageID{}, false
-	}
-}
-
-func toChunkedMessageID(msgID MessageID) (chunkMessageID, bool) {
-	cid, ok := msgID.(chunkMessageID)
-	if ok {
-		return cid, true
+		return nil
 	}
-	return chunkMessageID{}, false
 }
 
 func timeFromUnixTimestampMillis(timestamp uint64) time.Time {
@@ -449,16 +437,16 @@ func (t *ackTracker) toAckSet() []int64 {
 }
 
 type chunkMessageID struct {
-	messageID
+	*messageID
 
-	firstChunkID messageID
+	firstChunkID *messageID
 	receivedTime time.Time
 
 	consumer acker
 }
 
-func newChunkMessageID(firstChunkID messageID, lastChunkID messageID) chunkMessageID {
-	return chunkMessageID{
+func newChunkMessageID(firstChunkID *messageID, lastChunkID *messageID) *chunkMessageID {
+	return &chunkMessageID{
 		messageID:    lastChunkID,
 		firstChunkID: firstChunkID,
 		receivedTime: time.Now(),
diff --git a/pulsar/internal/channel_cond.go b/pulsar/internal/channel_cond.go
new file mode 100644
index 0000000..38301ab
--- /dev/null
+++ b/pulsar/internal/channel_cond.go
@@ -0,0 +1,76 @@
+// 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 internal
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"unsafe"
+)
+
+type chCond struct {
+	L sync.Locker
+	// The pointer to the channel, the channel pointed to may change,
+	// because we will use the channel's close mechanism to implement broadcast notifications.
+	notifyChPtr unsafe.Pointer
+}
+
+func newCond(l sync.Locker) *chCond {
+	c := &chCond{L: l}
+	n := make(chan struct{})
+	c.notifyChPtr = unsafe.Pointer(&n)
+	return c
+}
+
+// wait for broadcast calls. Similar to regular sync.Cond
+func (c *chCond) wait() {
+	n := c.notifyChan()
+	c.L.Unlock()
+	<-n
+	c.L.Lock()
+}
+
+// waitWithContext Same as wait() call, but the end condition can also be controlled through the context.
+func (c *chCond) waitWithContext(ctx context.Context) bool {
+	n := c.notifyChan()
+	c.L.Unlock()
+	defer c.L.Lock()
+	select {
+	case <-n:
+		return true
+	case <-ctx.Done():
+		return false
+	default:
+		return true
+	}
+}
+
+// broadcast wakes all goroutines waiting on c.
+// It is not required for the caller to hold c.L during the call.
+func (c *chCond) broadcast() {
+	n := make(chan struct{})
+	ptrOld := atomic.SwapPointer(&c.notifyChPtr, unsafe.Pointer(&n))
+	// close old channels to trigger broadcast.
+	close(*(*chan struct{})(ptrOld))
+}
+
+func (c *chCond) notifyChan() <-chan struct{} {
+	ptr := atomic.LoadPointer(&c.notifyChPtr)
+	return *((*chan struct{})(ptr))
+}
diff --git a/pulsar/internal/commands.go b/pulsar/internal/commands.go
index 70acf39..00e075b 100644
--- a/pulsar/internal/commands.go
+++ b/pulsar/internal/commands.go
@@ -218,6 +218,17 @@ func baseCommand(cmdType pb.BaseCommand_Type, msg proto.Message) *pb.BaseCommand
 		cmd.GetOrCreateSchema = msg.(*pb.CommandGetOrCreateSchema)
 	case pb.BaseCommand_GET_SCHEMA:
 		cmd.GetSchema = msg.(*pb.CommandGetSchema)
+	case pb.BaseCommand_TC_CLIENT_CONNECT_REQUEST:
+		cmd.TcClientConnectRequest = msg.(*pb.CommandTcClientConnectRequest)
+	case pb.BaseCommand_NEW_TXN:
+		cmd.NewTxn = msg.(*pb.CommandNewTxn)
+	case pb.BaseCommand_ADD_PARTITION_TO_TXN:
+		cmd.AddPartitionToTxn = msg.(*pb.CommandAddPartitionToTxn)
+	case pb.BaseCommand_ADD_SUBSCRIPTION_TO_TXN:
+		cmd.AddSubscriptionToTxn = msg.(*pb.CommandAddSubscriptionToTxn)
+	case pb.BaseCommand_END_TXN:
+		cmd.EndTxn = msg.(*pb.CommandEndTxn)
+
 	default:
 		panic(fmt.Sprintf("Missing command type: %v", cmdType))
 	}
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index 67b6f32..55d04f3 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -170,6 +170,8 @@ type connection struct {
 	metrics        *Metrics
 
 	keepAliveInterval time.Duration
+
+	lastActive time.Time
 }
 
 // connectionOptions defines configurations for creating connection.
@@ -583,7 +585,16 @@ func (c *connection) internalReceivedCommand(cmd *pb.BaseCommand, headersAndPayl
 		c.handlePing()
 	case pb.BaseCommand_PONG:
 		c.handlePong()
-
+	case pb.BaseCommand_TC_CLIENT_CONNECT_RESPONSE:
+		c.handleResponse(cmd.TcClientConnectResponse.GetRequestId(), cmd)
+	case pb.BaseCommand_NEW_TXN_RESPONSE:
+		c.handleResponse(cmd.NewTxnResponse.GetRequestId(), cmd)
+	case pb.BaseCommand_ADD_PARTITION_TO_TXN_RESPONSE:
+		c.handleResponse(cmd.AddPartitionToTxnResponse.GetRequestId(), cmd)
+	case pb.BaseCommand_ADD_SUBSCRIPTION_TO_TXN_RESPONSE:
+		c.handleResponse(cmd.AddSubscriptionToTxnResponse.GetRequestId(), cmd)
+	case pb.BaseCommand_END_TXN_RESPONSE:
+		c.handleResponse(cmd.EndTxnResponse.GetRequestId(), cmd)
 	case pb.BaseCommand_ACTIVE_CONSUMER_CHANGE:
 		c.handleActiveConsumerChange(cmd.GetActiveConsumerChange())
 
@@ -918,6 +929,52 @@ func (c *connection) UnregisterListener(id uint64) {
 	delete(c.listeners, id)
 }
 
+func (c *connection) ResetLastActive() {
+	c.Lock()
+	defer c.Unlock()
+	c.lastActive = time.Now()
+}
+
+func (c *connection) isIdle() bool {
+	{
+		c.pendingLock.Lock()
+		defer c.pendingLock.Unlock()
+		if len(c.pendingReqs) != 0 {
+			return false
+		}
+	}
+
+	{
+		c.listenersLock.RLock()
+		defer c.listenersLock.RUnlock()
+		if len(c.listeners) != 0 {
+			return false
+		}
+	}
+
+	{
+		c.consumerHandlersLock.Lock()
+		defer c.consumerHandlersLock.Unlock()
+		if len(c.consumerHandlers) != 0 {
+			return false
+		}
+	}
+
+	if len(c.incomingRequestsCh) != 0 || len(c.writeRequestsCh) != 0 {
+		return false
+	}
+	return true
+}
+
+func (c *connection) CheckIdle(maxIdleTime time.Duration) bool {
+	// We don't need to lock here because this method should only be
+	// called in a single goroutine of the connectionPool
+	if !c.isIdle() {
+		c.lastActive = time.Now()
+	}
+	return time.Since(c.lastActive) > maxIdleTime
+}
+
 // Close closes the connection by
 // closing underlying socket connection and closeCh.
 // This also triggers callbacks to the ConnectionClosed listeners.
diff --git a/pulsar/internal/connection_pool.go b/pulsar/internal/connection_pool.go
index 4f6d656..bf1e297 100644
--- a/pulsar/internal/connection_pool.go
+++ b/pulsar/internal/connection_pool.go
@@ -47,6 +47,7 @@ type connectionPool struct {
 	maxConnectionsPerHost int32
 	roundRobinCnt         int32
 	keepAliveInterval     time.Duration
+	closeCh               chan struct{}
 
 	metrics *Metrics
 	log     log.Logger
@@ -60,8 +61,9 @@ func NewConnectionPool(
 	keepAliveInterval time.Duration,
 	maxConnectionsPerHost int,
 	logger log.Logger,
-	metrics *Metrics) ConnectionPool {
-	return &connectionPool{
+	metrics *Metrics,
+	connectionMaxIdleTime time.Duration) ConnectionPool {
+	p := &connectionPool{
 		connections:           make(map[string]*connection),
 		tlsOptions:            tlsOptions,
 		auth:                  auth,
@@ -70,7 +72,10 @@ func NewConnectionPool(
 		keepAliveInterval:     keepAliveInterval,
 		log:                   logger,
 		metrics:               metrics,
+		closeCh:               make(chan struct{}),
 	}
+	go p.checkAndCleanIdleConnections(connectionMaxIdleTime)
+	return p
 }
 
 func (p *connectionPool) GetConnection(logicalAddr *url.URL, physicalAddr *url.URL) (Connection, error) {
@@ -109,6 +114,7 @@ func (p *connectionPool) GetConnection(logicalAddr *url.URL, physicalAddr *url.U
 		p.Unlock()
 		conn.start()
 	} else {
+		conn.ResetLastActive()
 		// we already have a connection
 		p.Unlock()
 	}
@@ -119,6 +125,7 @@ func (p *connectionPool) GetConnection(logicalAddr *url.URL, physicalAddr *url.U
 
 func (p *connectionPool) Close() {
 	p.Lock()
+	close(p.closeCh)
 	for k, c := range p.connections {
 		delete(p.connections, k)
 		c.Close()
@@ -134,3 +141,25 @@ func (p *connectionPool) getMapKey(addr *url.URL) string {
 	idx := cnt % p.maxConnectionsPerHost
 	return fmt.Sprint(addr.Host, '-', idx)
 }
+
+func (p *connectionPool) checkAndCleanIdleConnections(maxIdleTime time.Duration) {
+	if maxIdleTime < 0 {
+		return
+	}
+	for {
+		select {
+		case <-p.closeCh:
+			return
+		case <-time.After(maxIdleTime):
+			p.Lock()
+			for k, c := range p.connections {
+				if c.CheckIdle(maxIdleTime) {
+					c.log.Debugf("Closed connection due to inactivity.")
+					delete(p.connections, k)
+					c.Close()
+				}
+			}
+			p.Unlock()
+		}
+	}
+}
diff --git a/pulsar/internal/helper.go b/pulsar/internal/helper.go
new file mode 100644
index 0000000..3bca1ee
--- /dev/null
+++ b/pulsar/internal/helper.go
@@ -0,0 +1,33 @@
+// 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 internal
+
+import "time"
+
+// These method should only be used by tests
+
+func StartCleanConnectionsTask(p *ConnectionPool, connectionMaxIdleTime time.Duration) {
+	go (*p).(*connectionPool).checkAndCleanIdleConnections(connectionMaxIdleTime)
+}
+
+func GetConnectionsCount(p *ConnectionPool) int {
+	pool := (*p).(*connectionPool)
+	pool.Lock()
+	defer pool.Unlock()
+	return len(pool.connections)
+}
diff --git a/pulsar/internal/memory_limit_controller.go b/pulsar/internal/memory_limit_controller.go
new file mode 100644
index 0000000..5bf8d59
--- /dev/null
+++ b/pulsar/internal/memory_limit_controller.go
@@ -0,0 +1,101 @@
+// 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 internal
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+)
+
+type MemoryLimitController interface {
+	ReserveMemory(ctx context.Context, size int64) bool
+	TryReserveMemory(size int64) bool
+	ForceReserveMemory(size int64)
+	ReleaseMemory(size int64)
+	CurrentUsage() int64
+	CurrentUsagePercent() float64
+	IsMemoryLimited() bool
+}
+
+type memoryLimitController struct {
+	limit        int64
+	chCond       *chCond
+	currentUsage int64
+}
+
+func NewMemoryLimitController(limit int64) MemoryLimitController {
+	mlc := &memoryLimitController{
+		limit:  limit,
+		chCond: newCond(&sync.Mutex{}),
+	}
+	return mlc
+}
+
+func (m *memoryLimitController) ReserveMemory(ctx context.Context, size int64) bool {
+	if !m.TryReserveMemory(size) {
+		m.chCond.L.Lock()
+		defer m.chCond.L.Unlock()
+
+		for !m.TryReserveMemory(size) {
+			if !m.chCond.waitWithContext(ctx) {
+				return false
+			}
+		}
+	}
+	return true
+}
+
+func (m *memoryLimitController) TryReserveMemory(size int64) bool {
+	for {
+		current := atomic.LoadInt64(&m.currentUsage)
+		newUsage := current + size
+
+		// This condition means we allowed one request to go over the limit.
+		if m.IsMemoryLimited() && current > m.limit {
+			return false
+		}
+
+		if atomic.CompareAndSwapInt64(&m.currentUsage, current, newUsage) {
+			return true
+		}
+	}
+}
+
+func (m *memoryLimitController) ForceReserveMemory(size int64) {
+	atomic.AddInt64(&m.currentUsage, size)
+}
+
+func (m *memoryLimitController) ReleaseMemory(size int64) {
+	newUsage := atomic.AddInt64(&m.currentUsage, -size)
+	if newUsage+size > m.limit && newUsage <= m.limit {
+		m.chCond.broadcast()
+	}
+}
+
+func (m *memoryLimitController) CurrentUsage() int64 {
+	return atomic.LoadInt64(&m.currentUsage)
+}
+
+func (m *memoryLimitController) CurrentUsagePercent() float64 {
+	return float64(atomic.LoadInt64(&m.currentUsage)) / float64(m.limit)
+}
+
+func (m *memoryLimitController) IsMemoryLimited() bool {
+	return m.limit > 0
+}
diff --git a/pulsar/message.go b/pulsar/message.go
index d37692b..c44957d 100644
--- a/pulsar/message.go
+++ b/pulsar/message.go
@@ -181,3 +181,42 @@ func EarliestMessageID() MessageID {
 func LatestMessageID() MessageID {
 	return latestMessageID
 }
+
+func messageIDCompare(lhs MessageID, rhs MessageID) int {
+	if lhs.LedgerID() < rhs.LedgerID() {
+		return -1
+	} else if lhs.LedgerID() > rhs.LedgerID() {
+		return 1
+	}
+	if lhs.EntryID() < rhs.EntryID() {
+		return -1
+	} else if lhs.EntryID() > rhs.EntryID() {
+		return 1
+	}
+	// When performing batch index ACK on a batched message whose batch size is N,
+	// the ACK order should be:
+	//   (ledger, entry, 0) -> (ledger, entry, 1) -> ... -> (ledger, entry, N-1) -> (ledger, entry)
+	// So we have to treat any MessageID with the batch index precedes the MessageID without the batch index
+	// if they are in the same entry.
+	if lhs.BatchIdx() < 0 && rhs.BatchIdx() < 0 {
+		return 0
+	} else if lhs.BatchIdx() >= 0 && rhs.BatchIdx() < 0 {
+		return -1
+	} else if lhs.BatchIdx() < 0 && rhs.BatchIdx() >= 0 {
+		return 1
+	}
+	if lhs.BatchIdx() < rhs.BatchIdx() {
+		return -1
+	} else if lhs.BatchIdx() > rhs.BatchIdx() {
+		return 1
+	}
+	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/negative_acks_tracker.go b/pulsar/negative_acks_tracker.go
index 79ed694..58f5676 100644
--- a/pulsar/negative_acks_tracker.go
+++ b/pulsar/negative_acks_tracker.go
@@ -65,7 +65,7 @@ func newNegativeAcksTracker(rc redeliveryConsumer, delay time.Duration,
 	return t
 }
 
-func (t *negativeAcksTracker) Add(msgID messageID) {
+func (t *negativeAcksTracker) Add(msgID *messageID) {
 	// Always clear up the batch index since we want to track the nack
 	// for the entire batch
 	batchMsgID := messageID{
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index 160693c..744df79 100644
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -51,13 +51,14 @@ const (
 )
 
 var (
-	errFailAddToBatch  = newError(AddToBatchFailed, "message add to batch failed")
-	errSendTimeout     = newError(TimeoutError, "message send timeout")
-	errSendQueueIsFull = newError(ProducerQueueIsFull, "producer send queue is full")
-	errContextExpired  = newError(TimeoutError, "message send context expired")
-	errMessageTooLarge = newError(MessageTooBig, "message size exceeds MaxMessageSize")
-	errMetaTooLarge    = newError(InvalidMessage, "message metadata size exceeds MaxMessageSize")
-	errProducerClosed  = newError(ProducerClosed, "producer already been closed")
+	errFailAddToBatch     = newError(AddToBatchFailed, "message add to batch failed")
+	errSendTimeout        = newError(TimeoutError, "message send timeout")
+	errSendQueueIsFull    = newError(ProducerQueueIsFull, "producer send queue is full")
+	errContextExpired     = newError(TimeoutError, "message send context expired")
+	errMessageTooLarge    = newError(MessageTooBig, "message size exceeds MaxMessageSize")
+	errMetaTooLarge       = newError(InvalidMessage, "message metadata size exceeds MaxMessageSize")
+	errProducerClosed     = newError(ProducerClosed, "producer already been closed")
+	errMemoryBufferIsFull = newError(ClientMemoryBufferIsFull, "client memory buffer is full")
 
 	buffersPool sync.Pool
 )
@@ -483,6 +484,7 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 
 	// read payload from message
 	uncompressedPayload := msg.Payload
+	uncompressedPayloadSize := int64(len(uncompressedPayload))
 
 	var schemaPayload []byte
 	var err error
@@ -494,14 +496,14 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 
 	// The block chan must be closed when returned with exception
 	defer request.stopBlock()
-	if !p.canAddToQueue(request) {
+	if !p.canAddToQueue(request, uncompressedPayloadSize) {
 		return
 	}
 
 	if p.options.DisableMultiSchema {
 		if msg.Schema != nil && p.options.Schema != nil &&
 			msg.Schema.GetSchemaInfo().hash() != p.options.Schema.GetSchemaInfo().hash() {
-			p.publishSemaphore.Release()
+			p.releaseSemaphoreAndMem(uncompressedPayloadSize)
 			request.callback(nil, request.msg, fmt.Errorf("msg schema can not match with producer schema"))
 			p.log.WithError(err).Errorf("The producer %s of the topic %s is disabled the `MultiSchema`", p.producerName, p.topic)
 			return
@@ -520,7 +522,7 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 		if uncompressedPayload == nil && schema != nil {
 			schemaPayload, err = schema.Encode(msg.Value)
 			if err != nil {
-				p.publishSemaphore.Release()
+				p.releaseSemaphoreAndMem(uncompressedPayloadSize)
 				request.callback(nil, request.msg, newError(SchemaFailure, err.Error()))
 				p.log.WithError(err).Errorf("Schema encode message failed %s", msg.Value)
 				return
@@ -536,7 +538,7 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 		if schemaVersion == nil {
 			schemaVersion, err = p.getOrCreateSchema(schema.GetSchemaInfo())
 			if err != nil {
-				p.publishSemaphore.Release()
+				p.releaseSemaphoreAndMem(uncompressedPayloadSize)
 				p.log.WithError(err).Error("get schema version fail")
 				request.callback(nil, request.msg, fmt.Errorf("get schema version fail, err: %w", err))
 				return
@@ -596,7 +598,7 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 
 	// if msg is too large and chunking is disabled
 	if checkSize > maxMessageSize && !p.options.EnableChunking {
-		p.publishSemaphore.Release()
+		p.releaseSemaphoreAndMem(uncompressedPayloadSize)
 		request.callback(nil, request.msg, errMessageTooLarge)
 		p.log.WithError(errMessageTooLarge).
 			WithField("size", checkSize).
@@ -615,7 +617,7 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 	} else {
 		payloadChunkSize = int(p._getConn().GetMaxMessageSize()) - proto.Size(mm)
 		if payloadChunkSize <= 0 {
-			p.publishSemaphore.Release()
+			p.releaseSemaphoreAndMem(uncompressedPayloadSize)
 			request.callback(nil, msg, errMetaTooLarge)
 			p.log.WithError(errMetaTooLarge).
 				WithField("metadata size", proto.Size(mm)).
@@ -663,7 +665,8 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 					chunkRecorder:    cr,
 				}
 				// the permit of first chunk has acquired
-				if chunkID != 0 && !p.canAddToQueue(nsr) {
+				if chunkID != 0 && !p.canAddToQueue(nsr, 0) {
+					p.releaseSemaphoreAndMem(uncompressedPayloadSize - int64(rhs))
 					return
 				}
 				p.internalSingleSend(mm, compressedPayload[lhs:rhs], nsr, uint32(maxMessageSize))
@@ -688,7 +691,7 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 			// 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 {
-				p.publishSemaphore.Release()
+				p.releaseSemaphoreAndMem(uncompressedPayloadSize)
 				request.callback(nil, request.msg, errFailAddToBatch)
 				p.log.WithField("size", uncompressedSize).
 					WithField("properties", msg.Properties).
@@ -797,7 +800,7 @@ func (p *partitionProducer) internalSingleSend(mm *pb.MessageMetadata,
 		maxMessageSize,
 	); err != nil {
 		request.callback(nil, request.msg, err)
-		p.publishSemaphore.Release()
+		p.releaseSemaphoreAndMem(int64(len(msg.Payload)))
 		p.log.WithError(err).Errorf("Single message serialize failed %s", msg.Value)
 		return
 	}
@@ -935,7 +938,7 @@ func (p *partitionProducer) failTimeoutMessages() {
 				sr := i.(*sendRequest)
 				if sr.msg != nil {
 					size := len(sr.msg.Payload)
-					p.publishSemaphore.Release()
+					p.releaseSemaphoreAndMem(int64(size))
 					p.metrics.MessagesPending.Dec()
 					p.metrics.BytesPending.Sub(float64(size))
 					p.metrics.PublishErrorsTimeout.Inc()
@@ -1139,8 +1142,7 @@ func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt)
 			sr := i.(*sendRequest)
 			if sr.msg != nil {
 				atomic.StoreInt64(&p.lastSequenceID, int64(pi.sequenceID))
-				p.publishSemaphore.Release()
-
+				p.releaseSemaphoreAndMem(int64(len(sr.msg.Payload)))
 				p.metrics.PublishLatency.Observe(float64(now-sr.publishTime.UnixNano()) / 1.0e9)
 				p.metrics.MessagesPublished.Inc()
 				p.metrics.MessagesPending.Dec()
@@ -1161,7 +1163,7 @@ func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt)
 				if sr.totalChunks > 1 {
 					if sr.chunkID == 0 {
 						sr.chunkRecorder.setFirstChunkID(
-							messageID{
+							&messageID{
 								int64(response.MessageId.GetLedgerId()),
 								int64(response.MessageId.GetEntryId()),
 								-1,
@@ -1170,7 +1172,7 @@ func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt)
 							})
 					} else if sr.chunkID == sr.totalChunks-1 {
 						sr.chunkRecorder.setLastChunkID(
-							messageID{
+							&messageID{
 								int64(response.MessageId.GetLedgerId()),
 								int64(response.MessageId.GetEntryId()),
 								-1,
@@ -1178,7 +1180,7 @@ func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt)
 								0,
 							})
 						// use chunkMsgID to set msgID
-						msgID = sr.chunkRecorder.chunkedMsgID
+						msgID = &sr.chunkRecorder.chunkedMsgID
 					}
 				}
 
@@ -1326,7 +1328,12 @@ func (p *partitionProducer) _getConn() internal.Connection {
 	return p.conn.Load().(internal.Connection)
 }
 
-func (p *partitionProducer) canAddToQueue(sr *sendRequest) bool {
+func (p *partitionProducer) releaseSemaphoreAndMem(size int64) {
+	p.publishSemaphore.Release()
+	p.client.memLimit.ReleaseMemory(size)
+}
+
+func (p *partitionProducer) canAddToQueue(sr *sendRequest, uncompressedPayloadSize int64) bool {
 	if p.options.DisableBlockIfQueueFull {
 		if !p.publishSemaphore.TryAcquire() {
 			if sr.callback != nil {
@@ -1334,9 +1341,24 @@ func (p *partitionProducer) canAddToQueue(sr *sendRequest) bool {
 			}
 			return false
 		}
-	} else if !p.publishSemaphore.Acquire(sr.ctx) {
-		sr.callback(nil, sr.msg, errContextExpired)
-		return false
+		if !p.client.memLimit.TryReserveMemory(uncompressedPayloadSize) {
+			p.publishSemaphore.Release()
+			if sr.callback != nil {
+				sr.callback(nil, sr.msg, errMemoryBufferIsFull)
+			}
+			return false
+		}
+
+	} else {
+		if !p.publishSemaphore.Acquire(sr.ctx) {
+			sr.callback(nil, sr.msg, errContextExpired)
+			return false
+		}
+		if !p.client.memLimit.ReserveMemory(sr.ctx, uncompressedPayloadSize) {
+			p.publishSemaphore.Release()
+			sr.callback(nil, sr.msg, errContextExpired)
+			return false
+		}
 	}
 	p.metrics.MessagesPending.Inc()
 	p.metrics.BytesPending.Add(float64(len(sr.msg.Payload)))
@@ -1353,11 +1375,11 @@ func newChunkRecorder() *chunkRecorder {
 	}
 }
 
-func (c *chunkRecorder) setFirstChunkID(msgID messageID) {
+func (c *chunkRecorder) setFirstChunkID(msgID *messageID) {
 	c.chunkedMsgID.firstChunkID = msgID
 }
 
-func (c *chunkRecorder) setLastChunkID(msgID messageID) {
+func (c *chunkRecorder) setLastChunkID(msgID *messageID) {
 	c.chunkedMsgID.messageID = msgID
 }
 
diff --git a/pulsar/reader_impl.go b/pulsar/reader_impl.go
index 079754b..68dd084 100644
--- a/pulsar/reader_impl.go
+++ b/pulsar/reader_impl.go
@@ -37,7 +37,7 @@ type reader struct {
 	client              *client
 	pc                  *partitionConsumer
 	messageCh           chan ConsumerMessage
-	lastMessageInBroker trackingMessageID
+	lastMessageInBroker *trackingMessageID
 	log                 log.Logger
 	metrics             *internal.LeveledMetrics
 }
@@ -51,8 +51,8 @@ func newReader(client *client, options ReaderOptions) (Reader, error) {
 		return nil, newError(InvalidConfiguration, "StartMessageID is required")
 	}
 
-	startMessageID, ok := toTrackingMessageID(options.StartMessageID)
-	if !ok {
+	startMessageID := toTrackingMessageID(options.StartMessageID)
+	if startMessageID == nil {
 		// a custom type satisfying MessageID may not be a messageID or trackingMessageID
 		// so re-create messageID using its data
 		deserMsgID, err := deserializeMessageID(options.StartMessageID.Serialize())
@@ -60,8 +60,8 @@ func newReader(client *client, options ReaderOptions) (Reader, error) {
 			return nil, err
 		}
 		// de-serialized MessageID is a messageID
-		startMessageID = trackingMessageID{
-			messageID:    deserMsgID.(messageID),
+		startMessageID = &trackingMessageID{
+			messageID:    deserMsgID.(*messageID),
 			receivedTime: time.Now(),
 		}
 	}
@@ -148,7 +148,7 @@ func (r *reader) Next(ctx context.Context) (Message, error) {
 			// Acknowledge message immediately because the reader is based on non-durable subscription. When it reconnects,
 			// it will specify the subscription position anyway
 			msgID := cm.Message.ID()
-			if mid, ok := toTrackingMessageID(msgID); ok {
+			if mid := toTrackingMessageID(msgID); mid != nil {
 				r.pc.lastDequeuedMsg = mid
 				r.pc.AckID(mid)
 				return cm.Message, nil
@@ -161,7 +161,7 @@ func (r *reader) Next(ctx context.Context) (Message, error) {
 }
 
 func (r *reader) HasNext() bool {
-	if !r.lastMessageInBroker.Undefined() && r.hasMoreMessages() {
+	if r.lastMessageInBroker != nil && r.hasMoreMessages() {
 		return true
 	}
 
@@ -180,7 +180,7 @@ func (r *reader) HasNext() bool {
 }
 
 func (r *reader) hasMoreMessages() bool {
-	if !r.pc.lastDequeuedMsg.Undefined() {
+	if r.pc.lastDequeuedMsg != nil {
 		return r.lastMessageInBroker.isEntryIDValid() && r.lastMessageInBroker.greater(r.pc.lastDequeuedMsg.messageID)
 	}
 
@@ -200,29 +200,29 @@ func (r *reader) Close() {
 	r.metrics.ReadersClosed.Inc()
 }
 
-func (r *reader) messageID(msgID MessageID) (trackingMessageID, bool) {
-	mid, ok := toTrackingMessageID(msgID)
-	if !ok {
+func (r *reader) messageID(msgID MessageID) *trackingMessageID {
+	mid := toTrackingMessageID(msgID)
+	if mid == nil {
 		r.log.Warnf("invalid message id type %T", msgID)
-		return trackingMessageID{}, false
+		return nil
 	}
 
 	partition := int(mid.partitionIdx)
 	// did we receive a valid partition index?
 	if partition < 0 {
 		r.log.Warnf("invalid partition index %d expected", partition)
-		return trackingMessageID{}, false
+		return nil
 	}
 
-	return mid, true
+	return mid
 }
 
 func (r *reader) Seek(msgID MessageID) error {
 	r.Lock()
 	defer r.Unlock()
 
-	mid, ok := r.messageID(msgID)
-	if !ok {
+	mid := r.messageID(msgID)
+	if mid == nil {
 		return nil
 	}
 
diff --git a/pulsar/transaction_coordinator_client.go b/pulsar/transaction_coordinator_client.go
new file mode 100644
index 0000000..82d1490
--- /dev/null
+++ b/pulsar/transaction_coordinator_client.go
@@ -0,0 +1,199 @@
+// 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"
+	"strconv"
+	"sync/atomic"
+	"time"
+
+	"github.com/apache/pulsar-client-go/pulsar/internal"
+	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
+	"github.com/apache/pulsar-client-go/pulsar/log"
+	"google.golang.org/protobuf/proto"
+)
+
+type transactionCoordinatorClient struct {
+	client    *client
+	cons      []internal.Connection
+	epoch     uint64
+	semaphore internal.Semaphore
+	log       log.Logger
+}
+
+// TransactionCoordinatorAssign is the transaction_impl coordinator topic which is used to look up the broker
+// where the TC located.
+const TransactionCoordinatorAssign = "persistent://pulsar/system/transaction_coordinator_assign"
+
+// newTransactionCoordinatorClientImpl init a transactionImpl coordinator client and
+// acquire connections with all transactionImpl coordinators.
+func newTransactionCoordinatorClientImpl(client *client) *transactionCoordinatorClient {
+	tc := &transactionCoordinatorClient{
+		client:    client,
+		semaphore: internal.NewSemaphore(1000),
+	}
+	tc.log = client.log.SubLogger(log.Fields{})
+	return tc
+}
+
+func (tc *transactionCoordinatorClient) start() error {
+	r, err := tc.client.lookupService.GetPartitionedTopicMetadata(TransactionCoordinatorAssign)
+	if err != nil {
+		return err
+	}
+	tc.cons = make([]internal.Connection, r.Partitions)
+
+	//Get connections with all transaction_impl coordinators which is synchronized
+	for i := 0; i < r.Partitions; i++ {
+		err := tc.grabConn(uint64(i))
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (tc *transactionCoordinatorClient) grabConn(partition uint64) error {
+	lr, err := tc.client.lookupService.Lookup(getTCAssignTopicName(partition))
+	if err != nil {
+		tc.log.WithError(err).Warn("Failed to lookup the transaction_impl " +
+			"coordinator assign topic [" + strconv.FormatUint(partition, 10) + "]")
+		return err
+	}
+
+	requestID := tc.client.rpcClient.NewRequestID()
+	cmdTCConnect := pb.CommandTcClientConnectRequest{
+		RequestId: proto.Uint64(requestID),
+		TcId:      proto.Uint64(partition),
+	}
+
+	res, err := tc.client.rpcClient.Request(lr.LogicalAddr, lr.PhysicalAddr, requestID,
+		pb.BaseCommand_TC_CLIENT_CONNECT_REQUEST, &cmdTCConnect)
+
+	if err != nil {
+		tc.log.WithError(err).Error("Failed to connect transaction_impl coordinator " +
+			strconv.FormatUint(partition, 10))
+		return err
+	}
+	tc.cons[partition] = res.Cnx
+	return nil
+}
+
+func (tc *transactionCoordinatorClient) close() {
+	for _, con := range tc.cons {
+		con.Close()
+	}
+}
+
+// newTransaction new a transactionImpl which can be used to guarantee exactly-once semantics.
+func (tc *transactionCoordinatorClient) newTransaction(timeout time.Duration) (*TxnID, error) {
+	if err := tc.canSendRequest(); err != nil {
+		return nil, err
+	}
+	requestID := tc.client.rpcClient.NewRequestID()
+	nextTcID := tc.nextTCNumber()
+	cmdNewTxn := &pb.CommandNewTxn{
+		RequestId:     proto.Uint64(requestID),
+		TcId:          proto.Uint64(nextTcID),
+		TxnTtlSeconds: proto.Uint64(uint64(timeout.Milliseconds())),
+	}
+
+	cnx, err := tc.client.rpcClient.RequestOnCnx(tc.cons[nextTcID], requestID, pb.BaseCommand_NEW_TXN, cmdNewTxn)
+	tc.semaphore.Release()
+	if err != nil {
+		return nil, err
+	}
+
+	return &TxnID{*cnx.Response.NewTxnResponse.TxnidMostBits,
+		*cnx.Response.NewTxnResponse.TxnidLeastBits}, nil
+}
+
+// addPublishPartitionToTxn register the partitions which published messages with the transactionImpl.
+// And this can be used when ending the transactionImpl.
+func (tc *transactionCoordinatorClient) addPublishPartitionToTxn(id *TxnID, partitions []string) error {
+	if err := tc.canSendRequest(); err != nil {
+		return err
+	}
+	requestID := tc.client.rpcClient.NewRequestID()
+	cmdAddPartitions := &pb.CommandAddPartitionToTxn{
+		RequestId:      proto.Uint64(requestID),
+		TxnidMostBits:  proto.Uint64(id.mostSigBits),
+		TxnidLeastBits: proto.Uint64(id.leastSigBits),
+		Partitions:     partitions,
+	}
+	_, err := tc.client.rpcClient.RequestOnCnx(tc.cons[id.mostSigBits], requestID,
+		pb.BaseCommand_ADD_PARTITION_TO_TXN, cmdAddPartitions)
+	tc.semaphore.Release()
+	return err
+}
+
+// addSubscriptionToTxn register the subscription which acked messages with the transactionImpl.
+// And this can be used when ending the transactionImpl.
+func (tc *transactionCoordinatorClient) addSubscriptionToTxn(id *TxnID, topic string, subscription string) error {
+	if err := tc.canSendRequest(); err != nil {
+		return err
+	}
+	requestID := tc.client.rpcClient.NewRequestID()
+	sub := &pb.Subscription{
+		Topic:        &topic,
+		Subscription: &subscription,
+	}
+	cmdAddSubscription := &pb.CommandAddSubscriptionToTxn{
+		RequestId:      proto.Uint64(requestID),
+		TxnidMostBits:  proto.Uint64(id.mostSigBits),
+		TxnidLeastBits: proto.Uint64(id.leastSigBits),
+		Subscription:   []*pb.Subscription{sub},
+	}
+	_, err := tc.client.rpcClient.RequestOnCnx(tc.cons[id.mostSigBits], requestID,
+		pb.BaseCommand_ADD_SUBSCRIPTION_TO_TXN, cmdAddSubscription)
+	tc.semaphore.Release()
+	return err
+}
+
+// endTxn commit or abort the transactionImpl.
+func (tc *transactionCoordinatorClient) endTxn(id *TxnID, action pb.TxnAction) error {
+	if err := tc.canSendRequest(); err != nil {
+		return err
+	}
+	requestID := tc.client.rpcClient.NewRequestID()
+	cmdEndTxn := &pb.CommandEndTxn{
+		RequestId:      proto.Uint64(requestID),
+		TxnAction:      &action,
+		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)
+	tc.semaphore.Release()
+	return err
+}
+
+func getTCAssignTopicName(partition uint64) string {
+	return TransactionCoordinatorAssign + "-partition-" + strconv.FormatUint(partition, 10)
+}
+
+func (tc *transactionCoordinatorClient) canSendRequest() error {
+	if !tc.semaphore.Acquire(context.Background()) {
+		return newError(UnknownError, "Failed to acquire semaphore")
+	}
+	return nil
+}
+
+func (tc *transactionCoordinatorClient) nextTCNumber() uint64 {
+	return atomic.AddUint64(&tc.epoch, 1) % uint64(len(tc.cons))
+}


Test patch:
diff --git a/integration-tests/conf/standalone.conf b/integration-tests/conf/standalone.conf
index b9ff87b..c816c8f 100644
--- a/integration-tests/conf/standalone.conf
+++ b/integration-tests/conf/standalone.conf
@@ -96,8 +96,7 @@ anonymousUserRole=anonymous
 
 # Enable authentication
 authenticationEnabled=true
-
-# Autentication provider name list, which is comma separated list of class names
+# Authentication provider name list, which is comma separated list of class names
 authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderTls,org.apache.pulsar.broker.authentication.AuthenticationProviderToken,org.apache.pulsar.broker.authentication.AuthenticationProviderBasic
 
 # Enforce authorization
@@ -111,8 +110,10 @@ superUserRoles=localhost,superUser,admin
 
 # Authentication settings of the broker itself. Used when the broker connects to other brokers,
 # either in same or other clusters
-brokerClientAuthenticationPlugin=
-brokerClientAuthenticationParameters=
+brokerClientTlsEnabled=true
+brokerClientTrustCertsFilePath=/pulsar/certs/cacert.pem
+brokerClientAuthenticationPlugin=org.apache.pulsar.client.impl.auth.AuthenticationTls
+brokerClientAuthenticationParameters={"tlsCertFile":"/pulsar/certs/client-cert.pem","tlsKeyFile":"/pulsar/certs/client-key.pem"}
 
 ### --- BookKeeper Client --- ###
 
@@ -294,3 +295,23 @@ globalZookeeperServers=
 brokerServicePurgeInactiveFrequencyInSeconds=60
 
 acknowledgmentAtBatchIndexLevelEnabled=true
+### --- Transaction config variables --- ###
+# Enable transaction coordinator in broker
+transactionCoordinatorEnabled=true
+transactionMetadataStoreProviderClassName=org.apache.pulsar.transaction.coordinator.impl.MLTransactionMetadataStoreProvider
+
+# Transaction buffer take snapshot transaction count
+transactionBufferSnapshotMaxTransactionCount=1000
+
+# Transaction buffer take snapshot interval time
+# Unit : millisecond
+transactionBufferSnapshotMinTimeInMillis=5000
+
+# Enable or disable system topic
+systemTopicEnabled=true
+
+# The schema compatibility strategy is used for system topics.
+# Available values: ALWAYS_INCOMPATIBLE, ALWAYS_COMPATIBLE, BACKWARD, FORWARD, FULL, BACKWARD_TRANSITIVE, FORWARD_TRANSITIVE, FULL_TRANSITIVE
+systemTopicSchemaCompatibilityStrategy=ALWAYS_COMPATIBLE
+
+subscriptionKeySharedUseConsistentHashing=false
diff --git a/pulsar/ack_grouping_tracker_test.go b/pulsar/ack_grouping_tracker_test.go
new file mode 100644
index 0000000..e7a6725
--- /dev/null
+++ b/pulsar/ack_grouping_tracker_test.go
@@ -0,0 +1,197 @@
+// 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 (
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNoCacheTracker(t *testing.T) {
+	tests := []AckGroupingOptions{
+		{
+			MaxSize: 0,
+			MaxTime: 10 * time.Hour,
+		},
+		{
+			MaxSize: 1,
+			MaxTime: 10 * time.Hour,
+		},
+	}
+	for _, option := range tests {
+		t.Run(fmt.Sprintf("TestAckImmediately_size_%v_time_%vs", option.MaxSize, option.MaxTime.Seconds()),
+			func(t *testing.T) {
+				ledgerID0 := int64(-1)
+				ledgerID1 := int64(-1)
+				tracker := newAckGroupingTracker(&option,
+					func(id MessageID) { ledgerID0 = id.LedgerID() },
+					func(id MessageID) { ledgerID1 = id.LedgerID() })
+
+				tracker.add(&messageID{ledgerID: 1})
+				assert.Equal(t, atomic.LoadInt64(&ledgerID0), int64(1))
+				tracker.addCumulative(&messageID{ledgerID: 2})
+				assert.Equal(t, atomic.LoadInt64(&ledgerID1), int64(2))
+			})
+	}
+}
+
+type mockAcker struct {
+	sync.Mutex
+	ledgerIDs          []int64
+	cumulativeLedgerID int64
+}
+
+func (a *mockAcker) ack(id MessageID) {
+	defer a.Unlock()
+	a.Lock()
+	a.ledgerIDs = append(a.ledgerIDs, id.LedgerID())
+}
+
+func (a *mockAcker) ackCumulative(id MessageID) {
+	atomic.StoreInt64(&a.cumulativeLedgerID, id.LedgerID())
+}
+
+func (a *mockAcker) getLedgerIDs() []int64 {
+	defer a.Unlock()
+	a.Lock()
+	return a.ledgerIDs
+}
+
+func (a *mockAcker) getCumulativeLedgerID() int64 {
+	return atomic.LoadInt64(&a.cumulativeLedgerID)
+}
+
+func (a *mockAcker) reset() {
+	a.ledgerIDs = make([]int64, 0)
+	a.cumulativeLedgerID = int64(0)
+}
+
+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.add(&messageID{ledgerID: 1})
+	tracker.add(&messageID{ledgerID: 2})
+	for i := 1; i <= 2; i++ {
+		assert.True(t, tracker.isDuplicate(&messageID{ledgerID: int64(i)}))
+	}
+	assert.Equal(t, 0, len(acker.getLedgerIDs()))
+	tracker.add(&messageID{ledgerID: 3})
+	assert.Eventually(t, func() bool { return len(acker.getLedgerIDs()) > 0 },
+		10*time.Millisecond, 2*time.Millisecond)
+	assert.Equal(t, []int64{1, 2, 3}, acker.getLedgerIDs())
+	for i := 1; i <= 3; i++ {
+		assert.False(t, tracker.isDuplicate(&messageID{ledgerID: int64(i)}))
+	}
+
+	tracker.add(&messageID{ledgerID: 4})
+	// 4 won't be added because the cache is not full
+	assert.Equal(t, []int64{1, 2, 3}, acker.getLedgerIDs())
+
+	assert.False(t, tracker.isDuplicate(&messageID{ledgerID: 5}))
+	tracker.addCumulative(&messageID{ledgerID: 5})
+	for i := 0; i <= 5; i++ {
+		assert.True(t, tracker.isDuplicate(&messageID{ledgerID: int64(i)}))
+	}
+	assert.Equal(t, int64(5), acker.getCumulativeLedgerID())
+	assert.False(t, tracker.isDuplicate(&messageID{ledgerID: int64(6)}))
+
+	tracker.flush()
+	assert.Eventually(t, func() bool { return len(acker.getLedgerIDs()) > 3 },
+		10*time.Millisecond, 2*time.Millisecond)
+	assert.Equal(t, []int64{1, 2, 3, 4}, acker.getLedgerIDs())
+}
+
+func TestTimedTrackerIndividualAck(t *testing.T) {
+	var acker mockAcker
+	// MaxSize: 1000, MaxTime: 100ms
+	tracker := newAckGroupingTracker(nil, func(id MessageID) { acker.ack(id) }, nil)
+
+	expected := make([]int64, 0)
+	for i := 0; i < 999; i++ {
+		tracker.add(&messageID{ledgerID: int64(i)})
+		expected = append(expected, int64(i))
+	}
+	assert.Equal(t, 0, len(acker.getLedgerIDs()))
+
+	// case 1: flush because the tracker timed out
+	assert.Eventually(t, func() bool { return len(acker.getLedgerIDs()) == 999 },
+		150*time.Millisecond, 10*time.Millisecond)
+	assert.Equal(t, expected, acker.getLedgerIDs())
+
+	// case 2: flush because cache is full
+	time.Sleep(50) // see case 3
+	acker.reset()
+	expected = append(expected, 999)
+	for i := 0; i < 1001; i++ {
+		tracker.add(&messageID{ledgerID: int64(i)})
+	}
+	assert.Equal(t, expected, acker.getLedgerIDs())
+
+	// case 3: flush will reset the timer
+	start := time.Now()
+	assert.Eventually(t, func() bool { return len(acker.getLedgerIDs()) > 1000 },
+		150*time.Millisecond, 10*time.Millisecond)
+	elapsed := time.Since(start)
+	assert.GreaterOrEqual(t, elapsed, int64(100), "elapsed", elapsed)
+	assert.Equal(t, append(expected, 1000), acker.getLedgerIDs())
+}
+
+func TestTimedTrackerCumulativeAck(t *testing.T) {
+	var acker mockAcker
+	// MaxTime is 100ms
+	tracker := newAckGroupingTracker(nil, nil, func(id MessageID) { acker.ackCumulative(id) })
+
+	// case 1: flush because of the timeout
+	tracker.addCumulative(&messageID{ledgerID: 1})
+	assert.NotEqual(t, int64(1), acker.getCumulativeLedgerID())
+	assert.Eventually(t, func() bool { return acker.getCumulativeLedgerID() == int64(1) },
+		150*time.Millisecond, 10*time.Millisecond)
+	assert.Equal(t, int64(1), acker.getCumulativeLedgerID())
+
+	// case 2: flush manually
+	tracker.addCumulative(&messageID{ledgerID: 2})
+	tracker.flush()
+	assert.Equal(t, int64(2), acker.getCumulativeLedgerID())
+
+	// case 3: older MessageID cannot be acknowledged
+	tracker.addCumulative(&messageID{ledgerID: 1})
+	tracker.flush()
+	assert.Equal(t, int64(2), acker.getCumulativeLedgerID())
+}
+
+func TestTimedTrackerIsDuplicate(t *testing.T) {
+	tracker := newAckGroupingTracker(nil, func(id MessageID) {}, func(id MessageID) {})
+
+	tracker.add(&messageID{batchIdx: 0, batchSize: 3})
+	tracker.add(&messageID{batchIdx: 2, batchSize: 3})
+	assert.True(t, tracker.isDuplicate(&messageID{batchIdx: 0, batchSize: 3}))
+	assert.False(t, tracker.isDuplicate(&messageID{batchIdx: 1, batchSize: 3}))
+	assert.True(t, tracker.isDuplicate(&messageID{batchIdx: 2, batchSize: 3}))
+
+	tracker.flush()
+	assert.False(t, tracker.isDuplicate(&messageID{batchIdx: 0, batchSize: 3}))
+	assert.False(t, tracker.isDuplicate(&messageID{batchIdx: 1, batchSize: 3}))
+	assert.False(t, tracker.isDuplicate(&messageID{batchIdx: 2, batchSize: 3}))
+}
diff --git a/pulsar/auth/athenz_test.go b/pulsar/auth/athenz_test.go
index 8541ca9..8911808 100644
--- a/pulsar/auth/athenz_test.go
+++ b/pulsar/auth/athenz_test.go
@@ -30,7 +30,11 @@ import (
 	"github.com/stretchr/testify/mock"
 )
 
-const tlsClientKeyPath = "../../integration-tests/certs/client-key.pem"
+const (
+	clientKeyPath  = "../../integration-tests/certs/client-key.pem"
+	clientCertPath = "../../integration-tests/certs/client-cert.pem"
+	caCertPath     = "../../integration-tests/certs/cacert.pem"
+)
 
 type MockTokenBuilder struct {
 	mock.Mock
@@ -69,7 +73,7 @@ func (m *MockRoleToken) RoleTokenValue() (string, error) {
 
 func MockZmsNewTokenBuilder(domain, name string, privateKeyPEM []byte, keyVersion string) (zms.TokenBuilder, error) {
 	// assertion
-	key, err := os.ReadFile(tlsClientKeyPath)
+	key, err := os.ReadFile(clientKeyPath)
 	if err != nil {
 		return nil, err
 	}
@@ -105,16 +109,35 @@ func MockZtsNewRoleToken(tok zms.Token, domain string, opts zts.RoleTokenOptions
 	return mockRoleToken
 }
 
+func MockZtsNewRoleTokenFromCert(certFile, keyFile, domain string, opts zts.RoleTokenOptions) zts.RoleToken {
+	// assertion
+	if certFile != clientCertPath ||
+		keyFile != clientKeyPath ||
+		domain != "pulsar.test.provider" ||
+		opts.BaseZTSURL != "http://localhost:9999/zts/v1" ||
+		opts.AuthHeader != "" ||
+		opts.CACert == nil {
+		return nil
+	}
+
+	mockRoleToken := new(MockRoleToken)
+	mockRoleToken.On("RoleTokenValue").Return("mockRoleTokenFromCert", nil)
+	return mockRoleToken
+}
+
 func TestAthenzAuth(t *testing.T) {
-	privateKey := "file://" + tlsClientKeyPath
+	privateKey := "file://" + clientKeyPath
 	provider := NewAuthenticationAthenz(
-		"pulsar.test.provider",
-		"pulsar.test.tenant",
-		"service",
-		privateKey,
-		"",
-		"",
-		"http://localhost:9999")
+		"pulsar.test.provider",  // providerDomain
+		"pulsar.test.tenant",    // tenantDomain
+		"service",               // tenantService
+		privateKey,              // privateKey
+		"",                      // keyID
+		"",                      // x509CertChain
+		"",                      // caCert
+		"",                      // principalHeader
+		"",                      // roleHeader
+		"http://localhost:9999") // ztsURL
 
 	// inject mock function
 	athenz := provider.(*athenzAuthProvider)
@@ -128,3 +151,105 @@ func TestAthenzAuth(t *testing.T) {
 	assert.Equal(t, []byte("mockRoleToken"), data)
 	assert.NoError(t, err)
 }
+
+func TestCopperArgos(t *testing.T) {
+	privateKey := "file://" + clientKeyPath
+	x509CertChain := "file://" + clientCertPath
+	caCert := "file://" + caCertPath
+
+	provider := NewAuthenticationAthenz(
+		"pulsar.test.provider",  // providerDomain
+		"",                      // tenantDomain
+		"",                      // tenantService
+		privateKey,              // privateKey
+		"",                      // keyID
+		x509CertChain,           // x509CertChain
+		caCert,                  // caCert
+		"",                      // principalHeader
+		"",                      // roleHeader
+		"http://localhost:9999") // ztsURL
+
+	// inject mock function
+	athenz := provider.(*athenzAuthProvider)
+	athenz.ztsNewRoleTokenFromCert = MockZtsNewRoleTokenFromCert
+
+	err := athenz.Init()
+	assert.NoError(t, err)
+
+	data, err := athenz.GetData()
+	assert.Equal(t, []byte("mockRoleTokenFromCert"), data)
+	assert.NoError(t, err)
+}
+
+func TestIllegalParams(t *testing.T) {
+	privateKey := "file://" + clientKeyPath
+	x509CertChain := "file://" + clientCertPath
+
+	provider := NewAuthenticationAthenz(
+		"pulsar.test.provider",  // providerDomain
+		"",                      // tenantDomain
+		"",                      // tenantService
+		"",                      // privateKey
+		"",                      // keyID
+		"",                      // x509CertChain
+		"",                      // caCert
+		"",                      // principalHeader
+		"",                      // roleHeader
+		"http://localhost:9999") // ztsURL
+	athenz := provider.(*athenzAuthProvider)
+
+	err := athenz.Init()
+	assert.Error(t, err, "Should fail due to missing privateKey parameter")
+	assert.Equal(t, "missing required parameters", err.Error())
+
+	provider = NewAuthenticationAthenz(
+		"pulsar.test.provider",  // providerDomain
+		"pulsar.test.tenant",    // tenantDomain
+		"",                      // tenantService
+		privateKey,              // privateKey
+		"",                      // keyID
+		"",                      // x509CertChain
+		"",                      // caCert
+		"",                      // principalHeader
+		"",                      // roleHeader
+		"http://localhost:9999") // ztsURL
+	athenz = provider.(*athenzAuthProvider)
+
+	err = athenz.Init()
+	assert.Error(t, err, "Should fail due to missing tenantService parameter")
+	assert.Equal(t, "missing required parameters", err.Error())
+
+	provider = NewAuthenticationAthenz(
+		"pulsar.test.provider",  // providerDomain
+		"",                      // tenantDomain
+		"",                      // tenantService
+		privateKey,              // privateKey
+		"",                      // keyID
+		"data:foo",              // x509CertChain
+		"",                      // caCert
+		"",                      // principalHeader
+		"",                      // roleHeader
+		"http://localhost:9999") // ztsURL
+	athenz = provider.(*athenzAuthProvider)
+
+	err = athenz.Init()
+	assert.Error(t, err, "Should fail due to incorrect x509CertChain scheme")
+	assert.Equal(t, "x509CertChain and privateKey must be specified as file paths", err.Error())
+
+	provider = NewAuthenticationAthenz(
+		"pulsar.test.provider",  // providerDomain
+		"",                      // tenantDomain
+		"",                      // tenantService
+		"data:bar",              // privateKey
+		"",                      // keyID
+		x509CertChain,           // x509CertChain
+		"",                      // caCert
+		"",                      // principalHeader
+		"",                      // roleHeader
+		"http://localhost:9999") // ztsURL
+	athenz = provider.(*athenzAuthProvider)
+
+	err = athenz.Init()
+	assert.Error(t, err, "Should fail due to incorrect privateKey scheme")
+	assert.Equal(t, "x509CertChain and privateKey must be specified as file paths", err.Error())
+}
diff --git a/pulsar/client_impl_test.go b/pulsar/client_impl_test.go
index 299203f..bb28371 100644
--- a/pulsar/client_impl_test.go
+++ b/pulsar/client_impl_test.go
@@ -21,9 +21,11 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	"log"
 	"net/http"
 	"net/http/httptest"
 	"os"
+	"strings"
 	"testing"
 	"time"
 
@@ -425,7 +427,7 @@ func TestNamespaceTopics(t *testing.T) {
 		t.Fatal(err)
 	}
 	topic2 := fmt.Sprintf("%s/topic-2", namespace)
-	if err := httpPut("admin/v2/persistent/"+topic2, namespace); err != nil {
+	if err := httpPut("admin/v2/persistent/"+topic2, nil); err != nil {
 		t.Fatal(err)
 	}
 	defer func() {
@@ -446,7 +448,13 @@ func TestNamespaceTopics(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	assert.Equal(t, 2, len(topics))
+	topicCount := 0
+	for _, value := range topics {
+		if !strings.Contains(value, "__transaction_buffer_snapshot") {
+			topicCount++
+		}
+	}
+	assert.Equal(t, 2, topicCount)
 
 	// add a non-persistent topic
 	topicName := fmt.Sprintf("non-persistent://%s/testNonPersistentTopic", namespace)
@@ -467,7 +475,13 @@ func TestNamespaceTopics(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	assert.Equal(t, 2, len(topics))
+	topicCount = 0
+	for _, value := range topics {
+		if !strings.Contains(value, "__transaction_buffer_snapshot") {
+			topicCount++
+		}
+	}
+	assert.Equal(t, 2, topicCount)
 }
 
 func TestNamespaceTopicsWebURL(t *testing.T) {
@@ -488,7 +502,7 @@ func TestNamespaceTopicsWebURL(t *testing.T) {
 		t.Fatal(err)
 	}
 	topic2 := fmt.Sprintf("%s/topic-2", namespace)
-	if err := httpPut("admin/v2/persistent/"+topic2, namespace); err != nil {
+	if err := httpPut("admin/v2/persistent/"+topic2, nil); err != nil {
 		t.Fatal(err)
 	}
 	defer func() {
@@ -509,7 +523,13 @@ func TestNamespaceTopicsWebURL(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	assert.Equal(t, 2, len(topics))
+	topicCount := 0
+	for _, value := range topics {
+		if !strings.Contains(value, "__transaction_buffer_snapshot") {
+			topicCount++
+		}
+	}
+	assert.Equal(t, 2, topicCount)
 
 	// add a non-persistent topic
 	topicName := fmt.Sprintf("non-persistent://%s/testNonPersistentTopic", namespace)
@@ -530,7 +550,13 @@ func TestNamespaceTopicsWebURL(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	assert.Equal(t, 2, len(topics))
+	topicCount = 0
+	for _, value := range topics {
+		if !strings.Contains(value, "__transaction_buffer_snapshot") {
+			topicCount++
+		}
+	}
+	assert.Equal(t, 2, topicCount)
 }
 
 func anonymousNamespacePolicy() map[string]interface{} {
@@ -1098,3 +1124,117 @@ func TestServiceUrlTLSWithTLSTransportWithBasicAuth(t *testing.T) {
 func TestWebServiceUrlTLSWithTLSTransportWithBasicAuth(t *testing.T) {
 	testTLSTransportWithBasicAuth(t, webServiceURLTLS)
 }
+
+func TestConfigureConnectionMaxIdleTime(t *testing.T) {
+	_, err := NewClient(ClientOptions{
+		URL:                   serviceURL,
+		ConnectionMaxIdleTime: 1 * time.Second,
+	})
+
+	assert.Error(t, err, "Should be failed when the connectionMaxIdleTime is less than minConnMaxIdleTime")
+
+	cli, err := NewClient(ClientOptions{
+		URL:                   serviceURL,
+		ConnectionMaxIdleTime: -1, // Disabled
+	})
+
+	assert.Nil(t, err)
+	cli.Close()
+
+	cli, err = NewClient(ClientOptions{
+		URL:                   serviceURL,
+		ConnectionMaxIdleTime: 60 * time.Second,
+	})
+
+	assert.Nil(t, err)
+	cli.Close()
+}
+
+func testSendAndReceive(t *testing.T, producer Producer, consumer Consumer) {
+	// send 10 messages
+	for i := 0; i < 10; i++ {
+		if _, err := producer.Send(context.Background(), &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("hello-%d", i)),
+		}); err != nil {
+			log.Fatal(err)
+		}
+	}
+
+	// receive 10 messages
+	for i := 0; i < 10; i++ {
+		msg, err := consumer.Receive(context.Background())
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		expectMsg := fmt.Sprintf("hello-%d", i)
+		assert.Equal(t, []byte(expectMsg), msg.Payload())
+		// ack message
+		err = consumer.Ack(msg)
+		if err != nil {
+			return
+		}
+	}
+}
+
+func TestAutoCloseIdleConnection(t *testing.T) {
+	cli, err := NewClient(ClientOptions{
+		URL:                   serviceURL,
+		ConnectionMaxIdleTime: -1, // Disable auto release connections first, we will enable it manually later
+	})
+
+	assert.Nil(t, err)
+
+	topic := "TestAutoCloseIdleConnection"
+
+	// create consumer
+	consumer1, err := cli.Subscribe(ConsumerOptions{
+		Topic:            topic,
+		SubscriptionName: "my-sub",
+	})
+	assert.Nil(t, err)
+
+	// create producer
+	producer1, err := cli.CreateProducer(ProducerOptions{
+		Topic:           topic,
+		DisableBatching: false,
+	})
+	assert.Nil(t, err)
+
+	testSendAndReceive(t, producer1, consumer1)
+
+	pool := cli.(*client).cnxPool
+
+	producer1.Close()
+	consumer1.Close()
+
+	assert.NotEqual(t, 0, internal.GetConnectionsCount(&pool))
+
+	internal.StartCleanConnectionsTask(&pool, 2*time.Second) // Enable auto idle connections release manually
+
+	time.Sleep(6 * time.Second) // Need to wait at least 3 * ConnectionMaxIdleTime
+
+	assert.Equal(t, 0, internal.GetConnectionsCount(&pool))
+
+	// create consumer
+	consumer2, err := cli.Subscribe(ConsumerOptions{
+		Topic:            topic,
+		SubscriptionName: "my-sub",
+	})
+	assert.Nil(t, err)
+
+	// create producer
+	producer2, err := cli.CreateProducer(ProducerOptions{
+		Topic:           topic,
+		DisableBatching: false,
+	})
+	assert.Nil(t, err)
+
+	// Ensure the client still works
+	testSendAndReceive(t, producer2, consumer2)
+
+	producer2.Close()
+	consumer2.Close()
+
+	cli.Close()
+}
diff --git a/pulsar/consumer_partition_test.go b/pulsar/consumer_partition_test.go
index fd50b70..16c4399 100644
--- a/pulsar/consumer_partition_test.go
+++ b/pulsar/consumer_partition_test.go
@@ -37,6 +37,8 @@ func TestSingleMessageIDNoAckTracker(t *testing.T) {
 		metrics:              newTestMetrics(),
 		decryptor:            crypto.NewNoopDecryptor(),
 	}
+	pc.ackGroupingTracker = newAckGroupingTracker(&AckGroupingOptions{MaxSize: 0},
+		func(id MessageID) { pc.sendIndividualAck(id) }, nil)
 
 	headersAndPayload := internal.NewBufferWrapper(rawCompatSingleMessage)
 	if err := pc.MessageReceived(nil, headersAndPayload); err != nil {
@@ -46,11 +48,11 @@ func TestSingleMessageIDNoAckTracker(t *testing.T) {
 	// ensure the tracker was set on the message id
 	messages := <-pc.queueCh
 	for _, m := range messages {
-		assert.Nil(t, m.ID().(trackingMessageID).tracker)
+		assert.Nil(t, m.ID().(*trackingMessageID).tracker)
 	}
 
 	// ack the message id
-	pc.AckID(messages[0].msgID.(trackingMessageID))
+	pc.AckID(messages[0].msgID.(*trackingMessageID))
 
 	select {
 	case <-eventsCh:
@@ -73,6 +75,8 @@ func TestBatchMessageIDNoAckTracker(t *testing.T) {
 		metrics:              newTestMetrics(),
 		decryptor:            crypto.NewNoopDecryptor(),
 	}
+	pc.ackGroupingTracker = newAckGroupingTracker(&AckGroupingOptions{MaxSize: 0},
+		func(id MessageID) { pc.sendIndividualAck(id) }, nil)
 
 	headersAndPayload := internal.NewBufferWrapper(rawBatchMessage1)
 	if err := pc.MessageReceived(nil, headersAndPayload); err != nil {
@@ -82,11 +86,12 @@ func TestBatchMessageIDNoAckTracker(t *testing.T) {
 	// ensure the tracker was set on the message id
 	messages := <-pc.queueCh
 	for _, m := range messages {
-		assert.Nil(t, m.ID().(trackingMessageID).tracker)
+		assert.Nil(t, m.ID().(*trackingMessageID).tracker)
 	}
 
 	// ack the message id
-	pc.AckID(messages[0].msgID.(trackingMessageID))
+	err := pc.AckID(messages[0].msgID.(*trackingMessageID))
+	assert.Nil(t, err)
 
 	select {
 	case <-eventsCh:
@@ -105,6 +110,8 @@ func TestBatchMessageIDWithAckTracker(t *testing.T) {
 		metrics:              newTestMetrics(),
 		decryptor:            crypto.NewNoopDecryptor(),
 	}
+	pc.ackGroupingTracker = newAckGroupingTracker(&AckGroupingOptions{MaxSize: 0},
+		func(id MessageID) { pc.sendIndividualAck(id) }, nil)
 
 	headersAndPayload := internal.NewBufferWrapper(rawBatchMessage10)
 	if err := pc.MessageReceived(nil, headersAndPayload); err != nil {
@@ -114,12 +121,13 @@ func TestBatchMessageIDWithAckTracker(t *testing.T) {
 	// ensure the tracker was set on the message id
 	messages := <-pc.queueCh
 	for _, m := range messages {
-		assert.NotNil(t, m.ID().(trackingMessageID).tracker)
+		assert.NotNil(t, m.ID().(*trackingMessageID).tracker)
 	}
 
 	// ack all message ids except the last one
 	for i := 0; i < 9; i++ {
-		pc.AckID(messages[i].msgID.(trackingMessageID))
+		err := pc.AckID(messages[i].msgID.(*trackingMessageID))
+		assert.Nil(t, err)
 	}
 
 	select {
@@ -129,7 +137,8 @@ func TestBatchMessageIDWithAckTracker(t *testing.T) {
 	}
 
 	// ack last message
-	pc.AckID(messages[9].msgID.(trackingMessageID))
+	err := pc.AckID(messages[9].msgID.(*trackingMessageID))
+	assert.Nil(t, err)
 
 	select {
 	case <-eventsCh:
diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
index 0eb7aae..21fa7d0 100644
--- a/pulsar/consumer_test.go
+++ b/pulsar/consumer_test.go
@@ -910,7 +910,7 @@ func TestConsumerNoBatchCumulativeAck(t *testing.T) {
 
 		if i == N/2-1 {
 			// cumulative acks the first half of messages
-			consumer.AckCumulative(msg)
+			assert.Nil(t, consumer.AckCumulative(msg))
 		}
 	}
 
@@ -931,7 +931,7 @@ func TestConsumerNoBatchCumulativeAck(t *testing.T) {
 		assert.Nil(t, err)
 		assert.Equal(t, fmt.Sprintf("msg-content-%d", i), string(msg.Payload()))
 
-		consumer.Ack(msg)
+		assert.Nil(t, consumer.Ack(msg))
 	}
 }
 
@@ -3853,37 +3853,38 @@ func TestAckWithMessageID(t *testing.T) {
 }
 
 func TestBatchIndexAck(t *testing.T) {
-	tests := []struct {
-		AckWithResponse bool
-		Cumulative      bool
-	}{
-		{
-			AckWithResponse: true,
-			Cumulative:      true,
-		},
-		{
-			AckWithResponse: true,
-			Cumulative:      false,
-		},
-		{
-			AckWithResponse: false,
-			Cumulative:      true,
-		},
-		{
-			AckWithResponse: false,
-			Cumulative:      false,
-		},
-	}
-	for _, params := range tests {
-		t.Run(fmt.Sprintf("TestBatchIndexAck_WithResponse_%v_Cumulative_%v",
-			params.AckWithResponse, params.Cumulative),
+	type config struct {
+		ackWithResponse    bool
+		cumulative         bool
+		ackGroupingOptions *AckGroupingOptions
+	}
+	configs := make([]config, 0)
+	for _, option := range []*AckGroupingOptions{
+		nil, // MaxSize: 1000, MaxTime: 10ms
+		{MaxSize: 0, MaxTime: 0},
+		{MaxSize: 1000, MaxTime: 0},
+	} {
+		configs = append(configs, config{true, true, option})
+		configs = append(configs, config{true, false, option})
+		configs = append(configs, config{false, true, option})
+		configs = append(configs, config{false, false, option})
+	}
+
+	for _, params := range configs {
+		option := params.ackGroupingOptions
+		if option == nil {
+			option = &AckGroupingOptions{1000, 10 * time.Millisecond}
+		}
+
+		t.Run(fmt.Sprintf("TestBatchIndexAck_WithResponse_%v_Cumulative_%v_AckGroupingOption_%v_%v",
+			params.ackWithResponse, params.cumulative, option.MaxSize, option.MaxTime.Milliseconds()),
 			func(t *testing.T) {
-				runBatchIndexAckTest(t, params.AckWithResponse, params.Cumulative)
+				runBatchIndexAckTest(t, params.ackWithResponse, params.cumulative, params.ackGroupingOptions)
 			})
 	}
 }
 
-func runBatchIndexAckTest(t *testing.T, ackWithResponse bool, cumulative bool) {
+func runBatchIndexAckTest(t *testing.T, ackWithResponse bool, cumulative bool, option *AckGroupingOptions) {
 	client, err := NewClient(ClientOptions{
 		URL: lookupURL,
 	})
@@ -3897,6 +3898,7 @@ func runBatchIndexAckTest(t *testing.T, ackWithResponse bool, cumulative bool) {
 			SubscriptionName:               "my-sub",
 			AckWithResponse:                ackWithResponse,
 			EnableBatchIndexAcknowledgment: true,
+			AckGroupingOptions:             option,
 		})
 		assert.Nil(t, err)
 		return consumer
diff --git a/pulsar/helper_for_test.go b/pulsar/helper_for_test.go
index a2888d0..eda2d72 100644
--- a/pulsar/helper_for_test.go
+++ b/pulsar/helper_for_test.go
@@ -157,6 +157,13 @@ func topicStats(topic string) (map[string]interface{}, error) {
 	return metadata, err
 }
 
+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)
+	err := httpGet(path, &metadata)
+	return metadata, err
+}
+
 func topicPath(topic string) string {
 	tn, _ := internal.ParseTopicName(topic)
 	idx := strings.LastIndex(tn.Name, "/")
diff --git a/pulsar/impl_message_test.go b/pulsar/impl_message_test.go
index 413a39f..6a21171 100644
--- a/pulsar/impl_message_test.go
+++ b/pulsar/impl_message_test.go
@@ -31,11 +31,11 @@ func TestMessageId(t *testing.T) {
 	assert.NoError(t, err)
 	assert.NotNil(t, id2)
 
-	assert.Equal(t, int64(1), id2.(messageID).ledgerID)
-	assert.Equal(t, int64(2), id2.(messageID).entryID)
-	assert.Equal(t, int32(3), id2.(messageID).batchIdx)
-	assert.Equal(t, int32(4), id2.(messageID).partitionIdx)
-	assert.Equal(t, int32(5), id2.(messageID).batchSize)
+	assert.Equal(t, int64(1), id2.(*messageID).ledgerID)
+	assert.Equal(t, int64(2), id2.(*messageID).entryID)
+	assert.Equal(t, int32(3), id2.(*messageID).batchIdx)
+	assert.Equal(t, int32(4), id2.(*messageID).partitionIdx)
+	assert.Equal(t, int32(5), id2.(*messageID).batchSize)
 
 	id, err = DeserializeMessageID(nil)
 	assert.Error(t, err)
@@ -110,7 +110,7 @@ func TestAckingMessageIDBatchOne(t *testing.T) {
 
 func TestAckingMessageIDBatchTwo(t *testing.T) {
 	tracker := newAckTracker(2)
-	ids := []trackingMessageID{
+	ids := []*trackingMessageID{
 		newTrackingMessageID(1, 1, 0, 0, 0, tracker),
 		newTrackingMessageID(1, 1, 1, 0, 0, tracker),
 	}
@@ -121,7 +121,7 @@ func TestAckingMessageIDBatchTwo(t *testing.T) {
 
 	// try reverse order
 	tracker = newAckTracker(2)
-	ids = []trackingMessageID{
+	ids = []*trackingMessageID{
 		newTrackingMessageID(1, 1, 0, 0, 0, tracker),
 		newTrackingMessageID(1, 1, 1, 0, 0, tracker),
 	}
diff --git a/pulsar/internal/channel_cond_test.go b/pulsar/internal/channel_cond_test.go
new file mode 100644
index 0000000..a73d44e
--- /dev/null
+++ b/pulsar/internal/channel_cond_test.go
@@ -0,0 +1,55 @@
+// 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 internal
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestChCond(t *testing.T) {
+	cond := newCond(&sync.Mutex{})
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		cond.L.Lock()
+		cond.wait()
+		cond.L.Unlock()
+		wg.Done()
+	}()
+	time.Sleep(10 * time.Millisecond)
+	cond.broadcast()
+	wg.Wait()
+}
+
+func TestChCondWithContext(t *testing.T) {
+	cond := newCond(&sync.Mutex{})
+	wg := sync.WaitGroup{}
+	ctx, cancel := context.WithCancel(context.Background())
+	wg.Add(1)
+	go func() {
+		cond.L.Lock()
+		cond.waitWithContext(ctx)
+		cond.L.Unlock()
+		wg.Done()
+	}()
+	cancel()
+	wg.Wait()
+}
diff --git a/pulsar/internal/memory_limit_controller_test.go b/pulsar/internal/memory_limit_controller_test.go
new file mode 100644
index 0000000..a62c6e6
--- /dev/null
+++ b/pulsar/internal/memory_limit_controller_test.go
@@ -0,0 +1,184 @@
+// 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 internal
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestLimit(t *testing.T) {
+
+	mlc := NewMemoryLimitController(100)
+
+	for i := 0; i < 101; i++ {
+		assert.True(t, mlc.TryReserveMemory(1))
+	}
+
+	assert.False(t, mlc.TryReserveMemory(1))
+	assert.Equal(t, int64(101), mlc.CurrentUsage())
+	assert.InDelta(t, 1.01, mlc.CurrentUsagePercent(), 0.000001)
+
+	mlc.ReleaseMemory(1)
+	assert.Equal(t, int64(100), mlc.CurrentUsage())
+	assert.InDelta(t, 1.0, mlc.CurrentUsagePercent(), 0.000001)
+
+	assert.True(t, mlc.TryReserveMemory(1))
+	assert.Equal(t, int64(101), mlc.CurrentUsage())
+
+	mlc.ForceReserveMemory(99)
+	assert.False(t, mlc.TryReserveMemory(1))
+	assert.Equal(t, int64(200), mlc.CurrentUsage())
+	assert.InDelta(t, 2.0, mlc.CurrentUsagePercent(), 0.000001)
+
+	mlc.ReleaseMemory(50)
+	assert.False(t, mlc.TryReserveMemory(1))
+	assert.Equal(t, int64(150), mlc.CurrentUsage())
+	assert.InDelta(t, 1.5, mlc.CurrentUsagePercent(), 0.000001)
+}
+
+func TestDisableLimit(t *testing.T) {
+	mlc := NewMemoryLimitController(-1)
+	assert.True(t, mlc.TryReserveMemory(1000000))
+	assert.True(t, mlc.ReserveMemory(context.Background(), 1000000))
+	mlc.ReleaseMemory(1000000)
+	assert.Equal(t, int64(1000000), mlc.CurrentUsage())
+}
+
+func TestMultiGoroutineTryReserveMem(t *testing.T) {
+	mlc := NewMemoryLimitController(10000)
+
+	// Multi goroutine try reserve memory.
+	wg := sync.WaitGroup{}
+
+	wg.Add(10)
+	for i := 0; i < 10; i++ {
+		go func() {
+			for i := 0; i < 1000; i++ {
+				assert.True(t, mlc.TryReserveMemory(1))
+			}
+			wg.Done()
+		}()
+	}
+	assert.True(t, mlc.TryReserveMemory(1))
+	wg.Wait()
+	assert.False(t, mlc.TryReserveMemory(1))
+	assert.Equal(t, int64(10001), mlc.CurrentUsage())
+	assert.InDelta(t, 1.0001, mlc.CurrentUsagePercent(), 0.000001)
+}
+
+func TestReserveWithContext(t *testing.T) {
+	mlc := NewMemoryLimitController(100)
+	assert.True(t, mlc.TryReserveMemory(101))
+	gorNum := 10
+
+	// Reserve ctx timeout
+	waitGroup := sync.WaitGroup{}
+	waitGroup.Add(gorNum)
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+	for i := 0; i < gorNum; i++ {
+		go func() {
+			assert.False(t, mlc.ReserveMemory(ctx, 1))
+			waitGroup.Done()
+		}()
+	}
+	waitGroup.Wait()
+	assert.Equal(t, int64(101), mlc.CurrentUsage())
+
+	// Reserve ctx cancel
+	waitGroup.Add(gorNum)
+	cancelCtx, cancel := context.WithCancel(context.Background())
+	for i := 0; i < gorNum; i++ {
+		go func() {
+			assert.False(t, mlc.ReserveMemory(cancelCtx, 1))
+			waitGroup.Done()
+		}()
+	}
+	cancel()
+	waitGroup.Wait()
+	assert.Equal(t, int64(101), mlc.CurrentUsage())
+}
+
+func TestBlocking(t *testing.T) {
+	mlc := NewMemoryLimitController(100)
+	assert.True(t, mlc.TryReserveMemory(101))
+	assert.Equal(t, int64(101), mlc.CurrentUsage())
+	assert.InDelta(t, 1.01, mlc.CurrentUsagePercent(), 0.000001)
+
+	gorNum := 10
+	chs := make([]chan int, gorNum)
+	for i := 0; i < gorNum; i++ {
+		chs[i] = make(chan int, 1)
+		go reserveMemory(mlc, chs[i])
+	}
+
+	// The threads are blocked since the quota is full
+	for i := 0; i < gorNum; i++ {
+		assert.False(t, awaitCh(chs[i]))
+	}
+	assert.Equal(t, int64(101), mlc.CurrentUsage())
+
+	mlc.ReleaseMemory(int64(gorNum))
+	for i := 0; i < gorNum; i++ {
+		assert.True(t, awaitCh(chs[i]))
+	}
+	assert.Equal(t, int64(101), mlc.CurrentUsage())
+}
+
+func TestStepRelease(t *testing.T) {
+	mlc := NewMemoryLimitController(100)
+	assert.True(t, mlc.TryReserveMemory(101))
+	assert.Equal(t, int64(101), mlc.CurrentUsage())
+	assert.InDelta(t, 1.01, mlc.CurrentUsagePercent(), 0.000001)
+
+	gorNum := 10
+	ch := make(chan int, 1)
+	for i := 0; i < gorNum; i++ {
+		go reserveMemory(mlc, ch)
+	}
+
+	// The threads are blocked since the quota is full
+	assert.False(t, awaitCh(ch))
+	assert.Equal(t, int64(101), mlc.CurrentUsage())
+
+	for i := 0; i < gorNum; i++ {
+		mlc.ReleaseMemory(1)
+		assert.True(t, awaitCh(ch))
+		assert.False(t, awaitCh(ch))
+	}
+	assert.Equal(t, int64(101), mlc.CurrentUsage())
+}
+
+func reserveMemory(mlc MemoryLimitController, ch chan int) {
+	mlc.ReserveMemory(context.Background(), 1)
+	ch <- 1
+}
+
+func awaitCh(ch chan int) bool {
+	select {
+	case <-ch:
+		return true
+	case <-time.After(100 * time.Millisecond):
+		return false
+	}
+}
diff --git a/pulsar/message_chunking_test.go b/pulsar/message_chunking_test.go
index b3d64af..aac87c7 100644
--- a/pulsar/message_chunking_test.go
+++ b/pulsar/message_chunking_test.go
@@ -454,10 +454,10 @@ func TestChunkSize(t *testing.T) {
 		})
 		assert.NoError(t, err)
 		if size <= payloadChunkSize {
-			_, ok := msgID.(messageID)
+			_, ok := msgID.(*messageID)
 			assert.Equal(t, true, ok)
 		} else {
-			_, ok := msgID.(chunkMessageID)
+			_, ok := msgID.(*chunkMessageID)
 			assert.Equal(t, true, ok)
 		}
 	}
diff --git a/pulsar/negative_acks_tracker_test.go b/pulsar/negative_acks_tracker_test.go
index 5faa947..12d33de 100644
--- a/pulsar/negative_acks_tracker_test.go
+++ b/pulsar/negative_acks_tracker_test.go
@@ -81,13 +81,13 @@ func TestNacksTracker(t *testing.T) {
 	nmc := newNackMockedConsumer(nil)
 	nacks := newNegativeAcksTracker(nmc, testNackDelay, nil, log.DefaultNopLogger())
 
-	nacks.Add(messageID{
+	nacks.Add(&messageID{
 		ledgerID: 1,
 		entryID:  1,
 		batchIdx: 1,
 	})
 
-	nacks.Add(messageID{
+	nacks.Add(&messageID{
 		ledgerID: 2,
 		entryID:  2,
 		batchIdx: 1,
@@ -114,25 +114,25 @@ func TestNacksWithBatchesTracker(t *testing.T) {
 	nmc := newNackMockedConsumer(nil)
 	nacks := newNegativeAcksTracker(nmc, testNackDelay, nil, log.DefaultNopLogger())
 
-	nacks.Add(messageID{
+	nacks.Add(&messageID{
 		ledgerID: 1,
 		entryID:  1,
 		batchIdx: 1,
 	})
 
-	nacks.Add(messageID{
+	nacks.Add(&messageID{
 		ledgerID: 1,
 		entryID:  1,
 		batchIdx: 2,
 	})
 
-	nacks.Add(messageID{
+	nacks.Add(&messageID{
 		ledgerID: 1,
 		entryID:  1,
 		batchIdx: 3,
 	})
 
-	nacks.Add(messageID{
+	nacks.Add(&messageID{
 		ledgerID: 2,
 		entryID:  2,
 		batchIdx: 1,
@@ -194,7 +194,7 @@ func (msg *mockMessage1) Payload() []byte {
 }
 
 func (msg *mockMessage1) ID() MessageID {
-	return messageID{
+	return &messageID{
 		ledgerID: 1,
 		entryID:  1,
 		batchIdx: 1,
@@ -270,7 +270,7 @@ func (msg *mockMessage2) Payload() []byte {
 }
 
 func (msg *mockMessage2) ID() MessageID {
-	return messageID{
+	return &messageID{
 		ledgerID: 2,
 		entryID:  2,
 		batchIdx: 1,
diff --git a/pulsar/producer_test.go b/pulsar/producer_test.go
index e69a14c..a6f5e39 100644
--- a/pulsar/producer_test.go
+++ b/pulsar/producer_test.go
@@ -325,7 +325,7 @@ func TestFlushInProducer(t *testing.T) {
 		assert.Nil(t, err)
 		msgCount++
 
-		msgID := msg.ID().(trackingMessageID)
+		msgID := msg.ID().(*trackingMessageID)
 		// Since messages are batched, they will be sharing the same ledgerId/entryId
 		if ledgerID == -1 {
 			ledgerID = msgID.ledgerID
@@ -742,7 +742,7 @@ func TestBatchDelayMessage(t *testing.T) {
 	var delayMsgID int64
 	ch := make(chan struct{}, 2)
 	producer.SendAsync(ctx, delayMsg, func(id MessageID, producerMessage *ProducerMessage, err error) {
-		atomic.StoreInt64(&delayMsgID, id.(messageID).entryID)
+		atomic.StoreInt64(&delayMsgID, id.(*messageID).entryID)
 		ch <- struct{}{}
 	})
 	delayMsgPublished := false
@@ -758,13 +758,13 @@ func TestBatchDelayMessage(t *testing.T) {
 	}
 	var noDelayMsgID int64
 	producer.SendAsync(ctx, noDelayMsg, func(id MessageID, producerMessage *ProducerMessage, err error) {
-		atomic.StoreInt64(&noDelayMsgID, id.(messageID).entryID)
+		atomic.StoreInt64(&noDelayMsgID, id.(*messageID).entryID)
 	})
 	for i := 0; i < 2; i++ {
 		msg, err := consumer.Receive(context.Background())
 		assert.Nil(t, err, "unexpected error occurred when recving message from topic")
 
-		switch msg.ID().(trackingMessageID).entryID {
+		switch msg.ID().(*trackingMessageID).entryID {
 		case atomic.LoadInt64(&noDelayMsgID):
 			assert.LessOrEqual(t, time.Since(msg.PublishTime()).Nanoseconds(), int64(batchingDelay*2))
 		case atomic.LoadInt64(&delayMsgID):
@@ -1779,3 +1779,129 @@ func TestWaitForExclusiveProducer(t *testing.T) {
 	producer1.Close()
 	wg.Wait()
 }
+
+func TestMemLimitRejectProducerMessages(t *testing.T) {
+
+	c, err := NewClient(ClientOptions{
+		URL:              serviceURL,
+		MemoryLimitBytes: 100 * 1024,
+	})
+	assert.NoError(t, err)
+	defer c.Close()
+
+	topicName := newTopicName()
+	producer1, _ := c.CreateProducer(ProducerOptions{
+		Topic:                   topicName,
+		DisableBlockIfQueueFull: true,
+		DisableBatching:         false,
+		BatchingMaxPublishDelay: 100 * time.Second,
+		SendTimeout:             2 * time.Second,
+	})
+
+	producer2, _ := c.CreateProducer(ProducerOptions{
+		Topic:                   topicName,
+		DisableBlockIfQueueFull: true,
+		DisableBatching:         false,
+		BatchingMaxPublishDelay: 100 * time.Second,
+		SendTimeout:             2 * time.Second,
+	})
+
+	n := 101
+	for i := 0; i < n/2; i++ {
+		producer1.SendAsync(context.Background(), &ProducerMessage{
+			Payload: make([]byte, 1024),
+		}, func(id MessageID, message *ProducerMessage, e error) {})
+
+		producer2.SendAsync(context.Background(), &ProducerMessage{
+			Payload: make([]byte, 1024),
+		}, func(id MessageID, message *ProducerMessage, e error) {})
+	}
+	// Last message in order to reach the limit
+	producer1.SendAsync(context.Background(), &ProducerMessage{
+		Payload: make([]byte, 1024),
+	}, func(id MessageID, message *ProducerMessage, e error) {})
+	time.Sleep(100 * time.Millisecond)
+	assert.Equal(t, int64(n*1024), c.(*client).memLimit.CurrentUsage())
+
+	_, err = producer1.Send(context.Background(), &ProducerMessage{
+		Payload: make([]byte, 1024),
+	})
+	assert.Error(t, err)
+	assert.ErrorContains(t, err, getResultStr(ClientMemoryBufferIsFull))
+
+	_, err = producer2.Send(context.Background(), &ProducerMessage{
+		Payload: make([]byte, 1024),
+	})
+	assert.Error(t, err)
+	assert.ErrorContains(t, err, getResultStr(ClientMemoryBufferIsFull))
+
+	// flush pending msg
+	err = producer1.Flush()
+	assert.NoError(t, err)
+	err = producer2.Flush()
+	assert.NoError(t, err)
+	assert.Equal(t, int64(0), c.(*client).memLimit.CurrentUsage())
+
+	_, err = producer1.Send(context.Background(), &ProducerMessage{
+		Payload: make([]byte, 1024),
+	})
+	assert.NoError(t, err)
+	_, err = producer2.Send(context.Background(), &ProducerMessage{
+		Payload: make([]byte, 1024),
+	})
+	assert.NoError(t, err)
+}
+
+func TestMemLimitContextCancel(t *testing.T) {
+
+	c, err := NewClient(ClientOptions{
+		URL:              serviceURL,
+		MemoryLimitBytes: 100 * 1024,
+	})
+	assert.NoError(t, err)
+	defer c.Close()
+
+	topicName := newTopicName()
+	producer, _ := c.CreateProducer(ProducerOptions{
+		Topic:                   topicName,
+		DisableBlockIfQueueFull: false,
+		DisableBatching:         false,
+		BatchingMaxPublishDelay: 100 * time.Second,
+		SendTimeout:             2 * time.Second,
+	})
+
+	n := 101
+	ctx, cancel := context.WithCancel(context.Background())
+	for i := 0; i < n; i++ {
+		producer.SendAsync(ctx, &ProducerMessage{
+			Payload: make([]byte, 1024),
+		}, func(id MessageID, message *ProducerMessage, e error) {})
+	}
+	time.Sleep(100 * time.Millisecond)
+	assert.Equal(t, int64(n*1024), c.(*client).memLimit.CurrentUsage())
+
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		producer.SendAsync(ctx, &ProducerMessage{
+			Payload: make([]byte, 1024),
+		}, func(id MessageID, message *ProducerMessage, e error) {
+			assert.Error(t, e)
+			assert.ErrorContains(t, e, getResultStr(TimeoutError))
+			wg.Done()
+		})
+	}()
+
+	// cancel pending msg
+	cancel()
+	wg.Wait()
+
+	err = producer.Flush()
+	assert.NoError(t, err)
+	assert.Equal(t, int64(0), c.(*client).memLimit.CurrentUsage())
+
+	_, err = producer.Send(context.Background(), &ProducerMessage{
+		Payload: make([]byte, 1024),
+	})
+	assert.NoError(t, err)
+}
diff --git a/pulsar/impl_message_bench_test.go b/pulsar/transaction.go
similarity index 62%
rename from pulsar/impl_message_bench_test.go
rename to pulsar/transaction.go
index 4b6ca10..ae7c673 100644
--- a/pulsar/impl_message_bench_test.go
+++ b/pulsar/transaction.go
@@ -17,33 +17,7 @@
 
 package pulsar
 
-import (
-	"testing"
-)
-
-var (
-	usedByProducer messageID
-	usedByConsumer trackingMessageID
-)
-
-func producerCall(id messageID) messageID {
-	id.entryID++
-	return id
-}
-
-func consumerCall(id trackingMessageID) trackingMessageID {
-	id.entryID++
-	return id
-}
-
-func BenchmarkProducerCall(b *testing.B) {
-	for i := 0; i < b.N; i++ {
-		usedByProducer = producerCall(usedByProducer)
-	}
-}
-
-func BenchmarkConsumerCall(b *testing.B) {
-	for i := 0; i < b.N; i++ {
-		usedByConsumer = consumerCall(usedByConsumer)
-	}
+type TxnID struct {
+	mostSigBits  uint64
+	leastSigBits uint64
 }
diff --git a/pulsar/transaction_test.go b/pulsar/transaction_test.go
new file mode 100644
index 0000000..14a652f
--- /dev/null
+++ b/pulsar/transaction_test.go
@@ -0,0 +1,98 @@
+// 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 (
+	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
+	"github.com/stretchr/testify/assert"
+
+	"testing"
+	"time"
+)
+
+func TestTCClient(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.
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:            topic,
+		SubscriptionName: sub,
+	})
+	assert.NoError(t, err)
+	//3. Test newTransaction, addSubscriptionToTxn, addPublishPartitionToTxn
+	//Create a transaction1 and add subscription and publish topic to the transaction.
+	id1, err := tc.newTransaction(3 * time.Minute)
+	assert.NoError(t, err)
+	err = tc.addSubscriptionToTxn(id1, topic, sub)
+	assert.NoError(t, err)
+	err = tc.addPublishPartitionToTxn(id1, []string{topic})
+	assert.NoError(t, err)
+	//4. Verify the transaction1 stats
+	stats, err := transactionStats(id1)
+	assert.NoError(t, err)
+	assert.Equal(t, "OPEN", stats["status"])
+	producedPartitions := stats["producedPartitions"].(map[string]interface{})
+	ackedPartitions := stats["ackedPartitions"].(map[string]interface{})
+	_, ok := producedPartitions[topic]
+	assert.True(t, ok)
+	temp, ok := ackedPartitions[topic]
+	assert.True(t, ok)
+	subscriptions := temp.(map[string]interface{})
+	_, ok = subscriptions[sub]
+	assert.True(t, ok)
+	//5. Test End transaction
+	//Create transaction2 and Commit the transaction.
+	id2, err := tc.newTransaction(3 * time.Minute)
+	assert.NoError(t, err)
+	//6. Verify the transaction2 stats
+	stats2, err := transactionStats(id2)
+	assert.NoError(t, err)
+	assert.Equal(t, "OPEN", stats2["status"])
+	err = tc.endTxn(id2, pb.TxnAction_COMMIT)
+	assert.NoError(t, err)
+	stats2, err = transactionStats(id2)
+	//The transaction will be removed from txnMeta. Therefore, it is expected that stats2 is zero
+	if err == nil {
+		assert.Equal(t, "COMMITTED", stats2["status"])
+	} else {
+		assert.Equal(t, err.Error(), "http error status code: 404")
+	}
+	defer consumer.Close()
+	defer tc.close()
+	defer client.Close()
+}
+
+// createTcClient Create a transaction coordinator client to send request
+func createTcClient(t *testing.T) (*transactionCoordinatorClient, *client) {
+	c, err := NewClient(ClientOptions{
+		URL:                   webServiceURLTLS,
+		TLSTrustCertsFilePath: caCertsPath,
+		Authentication:        NewAuthenticationTLS(tlsClientCertPath, tlsClientKeyPath),
+	})
+	if err != nil {
+		t.Fatalf("Failed to create client due to %s", err.Error())
+	}
+	tcClient := newTransactionCoordinatorClientImpl(c.(*client))
+	if err = tcClient.start(); err != nil {
+		t.Fatalf("Failed to start transaction coordinator due to %s", err.Error())
+	}
+
+	return tcClient, c.(*client)
+}
diff --git a/scripts/pulsar-test-service-start.sh b/scripts/pulsar-test-service-start.sh
index a7c4797..ed90b06 100755
--- a/scripts/pulsar-test-service-start.sh
+++ b/scripts/pulsar-test-service-start.sh
@@ -51,19 +51,7 @@ until curl http://localhost:8080/metrics > /dev/null 2>&1 ; do sleep 1; done
 
 echo "-- Pulsar service is ready -- Configure permissions"
 
-# Create "standalone" cluster
-$PULSAR_ADMIN clusters create \
-        standalone \
-        --url http://localhost:8080/ \
-        --url-secure https://localhost:8443/ \
-        --broker-url pulsar://localhost:6650/ \
-        --broker-url-secure pulsar+ssl://localhost:6651/
-
-# Create "public" tenant
-$PULSAR_ADMIN tenants create public -r "anonymous" -c "standalone"
-
-# Create "public/default" with no auth required
-$PULSAR_ADMIN namespaces create public/default
+$PULSAR_ADMIN tenants update public -r anonymous
 $PULSAR_ADMIN namespaces grant-permission public/default \
                         --actions produce,consume \
                         --role "anonymous"