You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by xi...@apache.org on 2023/01/29 11:10:11 UTC
[pulsar-client-go] branch xiangying/txn/feature_support created (now 7131f74)
This is an automated email from the ASF dual-hosted git repository.
xiangying pushed a change to branch xiangying/txn/feature_support
in repository https://gitbox.apache.org/repos/asf/pulsar-client-go.git
at 7131f74 Implement transactionCoordinatorClient
This branch includes the following new commits:
new 7131f74 Implement transactionCoordinatorClient
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[pulsar-client-go] 01/01: Implement transactionCoordinatorClient
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xiangying pushed a commit to branch xiangying/txn/feature_support
in repository https://gitbox.apache.org/repos/asf/pulsar-client-go.git
commit 7131f747a5bbb74efbe9bbcb1bc07465925b6197
Author: xiangying <19...@qq.com>
AuthorDate: Sun Jan 29 19:08:46 2023 +0800
Implement transactionCoordinatorClient
---
pulsar/client.go | 2 +
pulsar/client_impl.go | 9 ++
pulsar/error.go | 4 +
pulsar/transactionCoordinatorClient.go | 203 +++++++++++++++++++++++++++++++++
4 files changed, 218 insertions(+)
diff --git a/pulsar/client.go b/pulsar/client.go
index f4642c6..4935734 100644
--- a/pulsar/client.go
+++ b/pulsar/client.go
@@ -123,6 +123,8 @@ type ClientOptions struct {
// Add custom labels to all the metrics reported by this client instance
CustomMetricsLabels map[string]string
+
+ IsEnableTransaction bool
}
// Client represents a pulsar client
diff --git a/pulsar/client_impl.go b/pulsar/client_impl.go
index 09b72af..fe89f2e 100644
--- a/pulsar/client_impl.go
+++ b/pulsar/client_impl.go
@@ -40,6 +40,7 @@ type client struct {
handlers internal.ClientHandlers
lookupService internal.LookupService
metrics *internal.Metrics
+ tcClient *transactionCoordinatorClient
log log.Logger
}
@@ -147,6 +148,14 @@ func newClient(options ClientOptions) (Client, error) {
c.handlers = internal.NewClientHandlers()
+ if options.IsEnableTransaction {
+ 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..21e7335 100644
--- a/pulsar/error.go
+++ b/pulsar/error.go
@@ -103,6 +103,10 @@ const (
ProducerClosed
// SchemaFailure means the payload could not be encoded using the Schema
SchemaFailure
+
+ //Transaction related error
+
+ ReachMaxPendingOps
)
// Error implement error interface, composed of two parts: msg and result.
diff --git a/pulsar/transactionCoordinatorClient.go b/pulsar/transactionCoordinatorClient.go
new file mode 100644
index 0000000..a0a40ee
--- /dev/null
+++ b/pulsar/transactionCoordinatorClient.go
@@ -0,0 +1,203 @@
+package pulsar
+
+import (
+ "context"
+ "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"
+ "github.com/gogo/protobuf/proto"
+ "strconv"
+ "sync/atomic"
+ "time"
+)
+
+type transactionCoordinatorClient struct {
+ client *client
+ cons []internal.Connection
+ epoch uint64
+ semaphore internal.Semaphore
+ blockIfReachMaxPendingOps bool
+ //The number of transaction coordinators
+ tcNum uint64
+ log log.Logger
+}
+
+// TransactionCoordinatorAssign is the transaction coordinator topic which is used to look up the broker
+// where the TC located.
+const TransactionCoordinatorAssign = "persistent://pulsar/system/transaction_coordinator_assign"
+
+/*
+*
+Init a transaction coordinator client and acquire connections with all transaction coordinators.
+*/
+func newTransactionCoordinatorClientImpl(client *client) *transactionCoordinatorClient {
+ tc := &transactionCoordinatorClient{
+ client: client,
+ blockIfReachMaxPendingOps: true,
+ 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.tcNum = uint64(r.Partitions)
+ //Get connections with all transaction coordinators which is synchronized
+ for i := uint64(0); i < tc.tcNum; i++ {
+ err := tc.grabConn(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 " +
+ "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 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()
+ }
+}
+
+/*
+*
+New a transaction which can be used to guarantee exactly-once semantics.
+*/
+func (tc *transactionCoordinatorClient) newTransaction(timeout time.Duration) (TxnID, error) {
+ _, err := tc.canSendRequest()
+ if err != nil {
+ return TxnID{}, 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)
+ if err != nil {
+ return TxnID{}, err
+ }
+
+ return TxnID{nextTcId, *cnx.Response.NewTxnResponse.TxnidLeastBits}, nil
+}
+
+/*
+*
+Register the partitions which published messages with the transaction.
+And this can be used when ending the transaction.
+*/
+func (tc *transactionCoordinatorClient) addPublishPartitionToTxn(id TxnID, partitions []string) error {
+ _, err := tc.canSendRequest()
+ if 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)
+ return err
+}
+
+/*
+*
+Register the subscription which acked messages with the transaction.
+And this can be used when ending the transaction.
+*/
+func (tc *transactionCoordinatorClient) addSubscriptionToTxn(id TxnID, topic string, subscription string) error {
+ _, err := tc.canSendRequest()
+ if 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)
+ return err
+}
+
+/*
+*
+Commit or abort the transaction.
+*/
+func (tc *transactionCoordinatorClient) endTxn(id TxnID, action pb.TxnAction) error {
+ _, err := tc.canSendRequest()
+ if 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)
+ return err
+}
+
+func getTCAssignTopicName(partition uint64) string {
+ return TransactionCoordinatorAssign + "-partition-" + strconv.FormatUint(partition, 10)
+}
+
+func (tc *transactionCoordinatorClient) canSendRequest() (bool, error) {
+ if tc.blockIfReachMaxPendingOps {
+ if tc.semaphore.Acquire(context.Background()) {
+ return false, newError(UnknownError, "Failed to acquire semaphore")
+ }
+ } else {
+ if !tc.semaphore.TryAcquire() {
+ return false, newError(ReachMaxPendingOps, "transaction coordinator reach max pending ops")
+ }
+ }
+ return true, nil
+}
+
+func (tc *transactionCoordinatorClient) nextTCNumber() uint64 {
+ return atomic.AddUint64(&tc.epoch, 1) % tc.tcNum
+}