Repository: apache/pulsar-client-go
Pull request #967 (2023-02-26 06:28:06)

[perf] Support batch index ACK and max number of messages in batch
### Motivation

Currently, the perf tool does not support the maximum number of messages in batch for producer and batch index ACK for consumer.

### Modifications

- Add the `--batching-max-size` option to configure the maximum number of messages for producer
- To keep the code style consistent, change the `--batching-max-size` config from `int` to `uint`
- Add the `--enable-batch-index-ack` option to enable batch index ACK for consumer.


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/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/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/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 67b6f32..09623e5 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -583,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())
 
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..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/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/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/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"