Repository: apache/pulsar-client-go
Pull request #953 (2023-02-13 06:48:34)

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



Patch:
diff --git a/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/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/client.go b/pulsar/client.go
index d75c977..135d22b 100644
--- a/pulsar/client.go
+++ b/pulsar/client.go
@@ -142,6 +142,8 @@ type ClientOptions struct {
 	// Specify metric registerer used to register metrics.
 	// Default prometheus.DefaultRegisterer
 	MetricsRegisterer prometheus.Registerer
+
+	EnableTransaction bool
 }
 
 // Client represents a pulsar client
diff --git a/pulsar/client_impl.go b/pulsar/client_impl.go
index e794e3d..f444804 100644
--- a/pulsar/client_impl.go
+++ b/pulsar/client_impl.go
@@ -41,6 +41,7 @@ type client struct {
 	handlers      internal.ClientHandlers
 	lookupService internal.LookupService
 	metrics       *internal.Metrics
+	tcClient      *transactionCoordinatorClient
 
 	log log.Logger
 }
@@ -162,6 +163,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..8ee1822 100644
--- a/pulsar/consumer_impl.go
+++ b/pulsar/consumer_impl.go
@@ -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{
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index e723f8a..95a8d32 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -116,6 +116,7 @@ type partitionConsumerOpts struct {
 	// in failover mode, this callback will be called when consumer change
 	consumerEventListener ConsumerEventListener
 	enableBatchIndexAck   bool
+	ackGroupingOptions    *AckGroupingOptions
 }
 
 type ConsumerEventListener interface {
@@ -167,6 +168,7 @@ type partitionConsumer struct {
 
 	chunkedMsgCtxMap   *chunkedMsgCtxMap
 	unAckChunksTracker *unAckChunksTracker
+	ackGroupingTracker ackGroupingTracker
 }
 
 func (pc *partitionConsumer) ActiveConsumerChanged(isActive bool) {
@@ -310,6 +312,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,
@@ -467,30 +472,35 @@ func (pc *partitionConsumer) ackID(msgID MessageID, withResponse bool) error {
 		return errors.New("failed to convert trackingMessageID")
 	}
 
-	ackReq := new(ackRequest)
-	ackReq.doneCh = make(chan struct{})
-	ackReq.ackType = individualAck
 	if !trackingID.Undefined() && 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 {
@@ -524,14 +534,12 @@ func (pc *partitionConsumer) internalAckIDCumulative(msgID MessageID, withRespon
 		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,12 +548,13 @@ 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)
@@ -554,7 +563,20 @@ func (pc *partitionConsumer) internalAckIDCumulative(msgID MessageID, withRespon
 		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) {
@@ -631,6 +653,9 @@ func (pc *partitionConsumer) Close() {
 		return
 	}
 
+	// flush all pending ACK requests and terminate the timer goroutine
+	pc.ackGroupingTracker.close()
+
 	// close chunkedMsgCtxMap
 	pc.chunkedMsgCtxMap.Close()
 
@@ -658,6 +683,7 @@ func (pc *partitionConsumer) Seek(msgID MessageID) error {
 		return errors.New("unhandled messageID type")
 	}
 
+	pc.ackGroupingTracker.flushAndClean()
 	pc.eventsCh <- req
 
 	// wait for the request to complete
@@ -715,6 +741,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 +984,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 {
diff --git a/pulsar/error.go b/pulsar/error.go
index ead5cf9..ce366f5 100644
--- a/pulsar/error.go
+++ b/pulsar/error.go
@@ -103,6 +103,13 @@ 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
 )
 
 // Error implement error interface, composed of two parts: msg and result.
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 cb4af33..09623e5 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -523,8 +523,14 @@ func (c *connection) internalReceivedCommand(cmd *pb.BaseCommand, headersAndPayl
 		c.handleResponse(cmd.Success.GetRequestId(), cmd)
 
 	case pb.BaseCommand_PRODUCER_SUCCESS:
-		c.handleResponse(cmd.ProducerSuccess.GetRequestId(), cmd)
-
+		if !*cmd.ProducerSuccess.ProducerReady {
+			request, ok := c.findPendingRequest(cmd.ProducerSuccess.GetRequestId())
+			if ok {
+				request.callback(cmd, nil)
+			}
+		} else {
+			c.handleResponse(cmd.ProducerSuccess.GetRequestId(), cmd)
+		}
 	case pb.BaseCommand_PARTITIONED_METADATA_RESPONSE:
 		c.checkServerError(cmd.PartitionMetadataResponse.Error)
 		c.handleResponse(cmd.PartitionMetadataResponse.GetRequestId(), cmd)
@@ -577,7 +583,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())
 
@@ -748,6 +763,13 @@ func (c *connection) deletePendingRequest(requestID uint64) (*request, bool) {
 	return request, ok
 }
 
+func (c *connection) findPendingRequest(requestID uint64) (*request, bool) {
+	c.pendingLock.Lock()
+	defer c.pendingLock.Unlock()
+	request, ok := c.pendingReqs[requestID]
+	return request, ok
+}
+
 func (c *connection) failPendingRequests(err error) bool {
 	c.pendingLock.Lock()
 	defer c.pendingLock.Unlock()
diff --git a/pulsar/internal/rpc_client.go b/pulsar/internal/rpc_client.go
index 378ab4f..0ee8ca9 100644
--- a/pulsar/internal/rpc_client.go
+++ b/pulsar/internal/rpc_client.go
@@ -129,14 +129,24 @@ func (c *rpcClient) Request(logicalAddr *url.URL, physicalAddr *url.URL, request
 			Cnx:      cnx,
 			Response: response,
 		}, err}
-		close(ch)
 	})
 
-	select {
-	case res := <-ch:
-		return res.RPCResult, res.error
-	case <-time.After(c.requestTimeout):
-		return nil, ErrRequestTimeOut
+	timeoutCh := time.After(c.requestTimeout)
+	for {
+		select {
+		case res := <-ch:
+			// Ignoring producer not ready response.
+			// Continue to wait for the producer to create successfully
+			if res.error == nil && *res.RPCResult.Response.Type == pb.BaseCommand_PRODUCER_SUCCESS {
+				if !*res.RPCResult.Response.ProducerSuccess.ProducerReady {
+					timeoutCh = nil
+					break
+				}
+			}
+			return res.RPCResult, res.error
+		case <-timeoutCh:
+			return nil, ErrRequestTimeOut
+		}
 	}
 }
 
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/producer.go b/pulsar/producer.go
index d9ac34b..8fcb891 100644
--- a/pulsar/producer.go
+++ b/pulsar/producer.go
@@ -64,6 +64,9 @@ const (
 	// ProducerAccessModeExclusive is required exclusive access for producer.
 	// Fail immediately if there's already a producer connected.
 	ProducerAccessModeExclusive
+
+	// ProducerAccessModeWaitForExclusive is pending until producer can acquire exclusive access.
+	ProducerAccessModeWaitForExclusive
 )
 
 // TopicMetadata represents a topic metadata.
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index eece055..160693c 100644
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -1367,6 +1367,8 @@ func toProtoProducerAccessMode(accessMode ProducerAccessMode) pb.ProducerAccessM
 		return pb.ProducerAccessMode_Shared
 	case ProducerAccessModeExclusive:
 		return pb.ProducerAccessMode_Exclusive
+	case ProducerAccessModeWaitForExclusive:
+		return pb.ProducerAccessMode_WaitForExclusive
 	}
 
 	return pb.ProducerAccessMode_Shared
diff --git a/pulsar/transaction.go b/pulsar/transaction.go
new file mode 100644
index 0000000..ae7c673
--- /dev/null
+++ b/pulsar/transaction.go
@@ -0,0 +1,23 @@
+// 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
+
+type TxnID struct {
+	mostSigBits  uint64
+	leastSigBits uint64
+}
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..334fc87 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,21 @@ 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
\ No newline at end of file
diff --git a/pulsar/ack_grouping_tracker_test.go b/pulsar/ack_grouping_tracker_test.go
new file mode 100644
index 0000000..d7903e8
--- /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/client_impl_test.go b/pulsar/client_impl_test.go
index 299203f..b8efe9a 100644
--- a/pulsar/client_impl_test.go
+++ b/pulsar/client_impl_test.go
@@ -24,6 +24,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"os"
+	"strings"
 	"testing"
 	"time"
 
@@ -425,7 +426,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 +447,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 +474,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 +501,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 +522,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 +549,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{} {
diff --git a/pulsar/consumer_partition_test.go b/pulsar/consumer_partition_test.go
index fd50b70..b9a9a02 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 {
@@ -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 {
@@ -105,6 +109,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 {
diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
index 0eb7aae..de90c0e 100644
--- a/pulsar/consumer_test.go
+++ b/pulsar/consumer_test.go
@@ -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/producer_test.go b/pulsar/producer_test.go
index d7950eb..e69a14c 100644
--- a/pulsar/producer_test.go
+++ b/pulsar/producer_test.go
@@ -1737,3 +1737,45 @@ func TestExclusiveProducer(t *testing.T) {
 	assert.Error(t, err, "Producer should be failed")
 	assert.True(t, strings.Contains(err.Error(), "ProducerBusy"))
 }
+
+func TestWaitForExclusiveProducer(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: serviceURL,
+		// Set the request timeout is 200ms
+		OperationTimeout: 200 * time.Millisecond,
+	})
+	assert.NoError(t, err)
+	defer client.Close()
+
+	topicName := newTopicName()
+	producer1, err := client.CreateProducer(ProducerOptions{
+		Topic:              topicName,
+		ProducerAccessMode: ProducerAccessModeExclusive,
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, producer1)
+
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		producer2, err := client.CreateProducer(ProducerOptions{
+			Topic:              topicName,
+			ProducerAccessMode: ProducerAccessModeWaitForExclusive,
+		})
+		defer producer2.Close()
+		assert.NoError(t, err)
+		assert.NotNil(t, producer2)
+
+		id, err := producer2.Send(context.Background(), &ProducerMessage{
+			Payload: make([]byte, 1024),
+		})
+		assert.Nil(t, err)
+		assert.NotNil(t, id)
+		wg.Done()
+	}()
+	// Because set the request timeout is 200ms before.
+	// Here waite 300ms to cover wait for exclusive producer never timeout
+	time.Sleep(300 * time.Millisecond)
+	producer1.Close()
+	wg.Wait()
+}
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"