You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "liangyepianzhou (via GitHub)" <gi...@apache.org> on 2023/02/13 06:48:36 UTC
[GitHub] [pulsar-client-go] liangyepianzhou opened a new pull request, #953: [feat][txn]Implement transactionCoordinatorClient
liangyepianzhou opened a new pull request, #953:
URL: https://github.com/apache/pulsar-client-go/pull/953
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] liangyepianzhou commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1116701945
##########
pulsar/client_impl_test.go:
##########
@@ -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 {
Review Comment:
This request got a 400 response after enabling TLS.
I confirmed to @freeznet it is an error code.
So I change its logic to be like the logic of topic 1.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] nodece commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1115407936
##########
pulsar/transaction_coordinator_client.go:
##########
@@ -0,0 +1,223 @@
+// 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
+ blockIfReachMaxPendingOps bool
+ //The number of transactionImpl coordinators
+ tcNum uint64
+ 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"
+
+/*
+*
+Init a transactionImpl coordinator client and acquire connections with all transactionImpl 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)
+ tc.cons = make([]internal.Connection, tc.tcNum)
+
+ //Get connections with all transaction_impl 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_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()
+ }
+}
+
+/*
+*
+New a transactionImpl 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{*cnx.Response.NewTxnResponse.TxnidMostBits,
+ *cnx.Response.NewTxnResponse.TxnidLeastBits}, nil
+}
+
+/*
+*
+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 {
+ 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 transactionImpl.
+And this can be used when ending the transactionImpl.
+*/
+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 transactionImpl.
+*/
+func (tc *transactionCoordinatorClient) endTxn(id *TxnID, action pb.TxnAction) error {
Review Comment:
Ok.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] nodece commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1114647474
##########
pulsar/transaction_coordinator_client.go:
##########
@@ -0,0 +1,223 @@
+// 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
+ blockIfReachMaxPendingOps bool
+ //The number of transactionImpl coordinators
+ tcNum uint64
+ 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"
+
+/*
+*
+Init a transactionImpl coordinator client and acquire connections with all transactionImpl 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)
+ tc.cons = make([]internal.Connection, tc.tcNum)
+
+ //Get connections with all transaction_impl 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_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()
+ }
+}
+
+/*
+*
+New a transactionImpl 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{*cnx.Response.NewTxnResponse.TxnidMostBits,
+ *cnx.Response.NewTxnResponse.TxnidLeastBits}, nil
+}
+
+/*
+*
+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 {
+ 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 transactionImpl.
+And this can be used when ending the transactionImpl.
+*/
+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 transactionImpl.
+*/
+func (tc *transactionCoordinatorClient) endTxn(id *TxnID, action pb.TxnAction) error {
Review Comment:
Sounds good. But when using the `*TxnID`, I always want to check if it is `nil`, so I suggest you use `TxnID`, this mean is not `nil`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] BewareMyPower commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1116643769
##########
pulsar/transaction_test.go:
##########
@@ -0,0 +1,100 @@
+// 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()
+}
+
+/*
+Create a transaction coordinator client to send request
+*/
Review Comment:
```suggestion
// Create a transaction coordinator client to send request
```
##########
pulsar/client.go:
##########
@@ -142,6 +142,8 @@ type ClientOptions struct {
// Specify metric registerer used to register metrics.
// Default prometheus.DefaultRegisterer
MetricsRegisterer prometheus.Registerer
+
+ IsEnableTransaction bool
Review Comment:
Could you remove the `Is` prefix? See other options that enable some feature:
- producer.go: EnableChunking
- consumer.go: EnableBatchIndexAcknowledgment, RetryEnable, EnableDefaultNackBackoffPolicy
##########
pulsar/transaction_coordinator_client.go:
##########
@@ -0,0 +1,210 @@
+// 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
Review Comment:
It's never released. You should release the semaphore for each response received
##########
pulsar/transaction_coordinator_client.go:
##########
@@ -0,0 +1,210 @@
+// 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
+ blockIfReachMaxPendingOps bool
+ //The number of transactionImpl coordinators
+ tcNum uint64
+ 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,
+ 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)
+ tc.cons = make([]internal.Connection, tc.tcNum)
+
+ //Get connections with all transaction_impl 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_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) {
+ err := tc.canSendRequest()
+ if err != nil {
+ return nil, err
+ }
Review Comment:
```suggestion
if err := tc.canSendRequest(); err != nil {
return nil, err
}
```
Simplify the code
##########
pulsar/transaction_coordinator_client.go:
##########
@@ -0,0 +1,210 @@
+// 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
+ blockIfReachMaxPendingOps bool
Review Comment:
Could this field be false? I see `transactionCoordinatorClient` should only be created by `newTransactionCoordinatorClientImpl`, which set it true. If it's always true, please remove this field.
##########
pulsar/transaction_coordinator_client.go:
##########
@@ -0,0 +1,210 @@
+// 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
+ blockIfReachMaxPendingOps bool
+ //The number of transactionImpl coordinators
+ tcNum uint64
Review Comment:
This field is redundant. We can replace the use of it with `len(cons)`.
##########
pulsar/transaction_coordinator_client.go:
##########
@@ -0,0 +1,210 @@
+// 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
+ blockIfReachMaxPendingOps bool
+ //The number of transactionImpl coordinators
+ tcNum uint64
+ 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,
+ 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)
+ tc.cons = make([]internal.Connection, tc.tcNum)
+
+ //Get connections with all transaction_impl 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_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) {
+ err := tc.canSendRequest()
+ if 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)
+ 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 {
+ err := tc.canSendRequest()
+ if err != nil {
+ return err
+ }
Review Comment:
```suggestion
if err := tc.canSendRequest(); err != nil {
return nil, err
}
```
Simplify the code. The following `_, err = ...` in line 146 should become `_, err := ...` after this change.
##########
pulsar/client_impl_test.go:
##########
@@ -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 {
Review Comment:
Could you explain changes in this test?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] shibd commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1112724648
##########
pulsar/transactionCoordinatorClient.go:
##########
@@ -0,0 +1,222 @@
+// 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
+ blockIfReachMaxPendingOps bool
+ //The number of transactionImpl coordinators
+ tcNum uint64
+ 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"
+
+/*
+*
+Init a transactionImpl coordinator client and acquire connections with all transactionImpl 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)
+ tc.cons = make([]internal.Connection, tc.tcNum)
+
+ //Get connections with all transaction_impl 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_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()
+ }
+}
+
+/*
+*
+New a transactionImpl which can be used to guarantee exactly-once semantics.
+*/
+func (tc *transactionCoordinatorClient) newTransaction(timeout time.Duration) (TxnID, error) {
Review Comment:
This method will be changed to return `transaction` object in the future, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] liangyepianzhou commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1114391461
##########
pulsar/transaction_coordinator_client.go:
##########
@@ -0,0 +1,223 @@
+// 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
+ blockIfReachMaxPendingOps bool
+ //The number of transactionImpl coordinators
+ tcNum uint64
+ 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"
+
+/*
+*
+Init a transactionImpl coordinator client and acquire connections with all transactionImpl 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)
+ tc.cons = make([]internal.Connection, tc.tcNum)
+
+ //Get connections with all transaction_impl 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_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()
+ }
+}
+
+/*
+*
+New a transactionImpl 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{*cnx.Response.NewTxnResponse.TxnidMostBits,
+ *cnx.Response.NewTxnResponse.TxnidLeastBits}, nil
+}
+
+/*
+*
+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 {
+ 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 transactionImpl.
+And this can be used when ending the transactionImpl.
+*/
+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 transactionImpl.
+*/
+func (tc *transactionCoordinatorClient) endTxn(id *TxnID, action pb.TxnAction) error {
Review Comment:
Why should we use TxnID instead of * TxnID here? I understand that using *TxnID can reduce the memory copy during value transfer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] liangyepianzhou commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1114388239
##########
pulsar/transaction_coordinator_client.go:
##########
@@ -0,0 +1,223 @@
+// 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
+ blockIfReachMaxPendingOps bool
+ //The number of transactionImpl coordinators
+ tcNum uint64
+ 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"
+
+/*
+*
+Init a transactionImpl coordinator client and acquire connections with all transactionImpl 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)
+ tc.cons = make([]internal.Connection, tc.tcNum)
+
+ //Get connections with all transaction_impl 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_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()
+ }
+}
+
+/*
+*
+New a transactionImpl 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{*cnx.Response.NewTxnResponse.TxnidMostBits,
+ *cnx.Response.NewTxnResponse.TxnidLeastBits}, nil
+}
+
+/*
+*
+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 {
+ 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 transactionImpl.
+And this can be used when ending the transactionImpl.
+*/
+func (tc *transactionCoordinatorClient) addSubscriptionToTxn(id *TxnID, topic string, subscription string) error {
Review Comment:
Why should we use TxnID instead of * TxnID here? I understand that using `*TxnID` can reduce the memory copy during value transfer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] BewareMyPower commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1116431680
##########
pulsar/transaction_coordinator_client.go:
##########
@@ -0,0 +1,223 @@
+// 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
+ blockIfReachMaxPendingOps bool
+ //The number of transactionImpl coordinators
+ tcNum uint64
+ 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"
+
+/*
+*
+Init a transactionImpl coordinator client and acquire connections with all transactionImpl 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)
+ tc.cons = make([]internal.Connection, tc.tcNum)
+
+ //Get connections with all transaction_impl 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_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()
+ }
+}
+
+/*
+*
+New a transactionImpl 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 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)
+ if err != nil {
+ return nil, err
+ }
+
+ return &TxnID{*cnx.Response.NewTxnResponse.TxnidMostBits,
+ *cnx.Response.NewTxnResponse.TxnidLeastBits}, nil
+}
+
+/*
+*
+Register the partitions which published messages with the transactionImpl.
+And this can be used when ending the transactionImpl.
+*/
Review Comment:
Please make the comment style consistent. Use
```go
// line 1...
// line 2...
```
or
```go
/**
* line1
* line2
* ...
*/
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] nodece commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1114369185
##########
pulsar/transaction_coordinator_client.go:
##########
@@ -0,0 +1,223 @@
+// 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
+ blockIfReachMaxPendingOps bool
+ //The number of transactionImpl coordinators
+ tcNum uint64
+ 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"
+
+/*
+*
+Init a transactionImpl coordinator client and acquire connections with all transactionImpl 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)
+ tc.cons = make([]internal.Connection, tc.tcNum)
+
+ //Get connections with all transaction_impl 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_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()
+ }
+}
+
+/*
+*
+New a transactionImpl 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{*cnx.Response.NewTxnResponse.TxnidMostBits,
+ *cnx.Response.NewTxnResponse.TxnidLeastBits}, nil
+}
+
+/*
+*
+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 {
+ 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 transactionImpl.
+And this can be used when ending the transactionImpl.
+*/
+func (tc *transactionCoordinatorClient) addSubscriptionToTxn(id *TxnID, topic string, subscription string) error {
Review Comment:
```suggestion
func (tc *transactionCoordinatorClient) addSubscriptionToTxn(id TxnID, topic string, subscription string) error {
```
##########
pulsar/transaction_coordinator_client.go:
##########
@@ -0,0 +1,223 @@
+// 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
+ blockIfReachMaxPendingOps bool
+ //The number of transactionImpl coordinators
+ tcNum uint64
+ 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"
+
+/*
+*
+Init a transactionImpl coordinator client and acquire connections with all transactionImpl 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)
+ tc.cons = make([]internal.Connection, tc.tcNum)
+
+ //Get connections with all transaction_impl 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_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()
+ }
+}
+
+/*
+*
+New a transactionImpl 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{*cnx.Response.NewTxnResponse.TxnidMostBits,
+ *cnx.Response.NewTxnResponse.TxnidLeastBits}, nil
+}
+
+/*
+*
+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 {
Review Comment:
```suggestion
func (tc *transactionCoordinatorClient) addPublishPartitionToTxn(id TxnID, partitions []string) error {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] nodece commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1114099685
##########
pulsar/transactionCoordinatorClient.go:
##########
@@ -0,0 +1,223 @@
+// Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
`pulsar/transactionCoordinatorClient.go` rename to `pulsar/transaction_coordinator_client.go`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] nodece commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1114370212
##########
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) {
Review Comment:
```suggestion
func transactionStats(id TxnID) (map[string]interface{}, error) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] liangyepianzhou commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1116437391
##########
integration-tests/conf/standalone.conf:
##########
@@ -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"}
Review Comment:
Transaction coordinators use `brokerClient` to send commands to TB and TP.
So we need to add TLS authentication to `brokerClient` here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] nodece merged pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece merged PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] nodece commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1114060803
##########
Makefile:
##########
@@ -18,7 +18,7 @@
#
IMAGE_NAME = pulsar-client-go-test:latest
-PULSAR_VERSION ?= 2.8.3
+PULSAR_VERSION ?= 2.10.3
Review Comment:
Fixed version is right.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] nodece commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1114101702
##########
pulsar/transactionCoordinatorClient.go:
##########
@@ -0,0 +1,223 @@
+// 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
+ blockIfReachMaxPendingOps bool
+ //The number of transactionImpl coordinators
+ tcNum uint64
+ 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"
+
+/*
+*
+Init a transactionImpl coordinator client and acquire connections with all transactionImpl 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)
+ tc.cons = make([]internal.Connection, tc.tcNum)
+
+ //Get connections with all transaction_impl 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_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()
+ }
+}
+
+/*
+*
+New a transactionImpl which can be used to guarantee exactly-once semantics.
+*/
+func (tc *transactionCoordinatorClient) newTransaction(timeout time.Duration) (TxnID, error) {
Review Comment:
Rerturn the `*TxnID, error` is better.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] nodece commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1114368778
##########
pulsar/transaction_coordinator_client.go:
##########
@@ -0,0 +1,223 @@
+// 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
+ blockIfReachMaxPendingOps bool
+ //The number of transactionImpl coordinators
+ tcNum uint64
+ 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"
+
+/*
+*
+Init a transactionImpl coordinator client and acquire connections with all transactionImpl 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)
+ tc.cons = make([]internal.Connection, tc.tcNum)
+
+ //Get connections with all transaction_impl 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_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()
+ }
+}
+
+/*
+*
+New a transactionImpl 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{*cnx.Response.NewTxnResponse.TxnidMostBits,
+ *cnx.Response.NewTxnResponse.TxnidLeastBits}, nil
+}
+
+/*
+*
+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 {
+ 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 transactionImpl.
+And this can be used when ending the transactionImpl.
+*/
+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 transactionImpl.
+*/
+func (tc *transactionCoordinatorClient) endTxn(id *TxnID, action pb.TxnAction) error {
Review Comment:
```suggestion
func (tc *transactionCoordinatorClient) endTxn(id TxnID, action pb.TxnAction) error {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] nodece commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1114368240
##########
pulsar/transaction_coordinator_client.go:
##########
@@ -0,0 +1,223 @@
+// 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
+ blockIfReachMaxPendingOps bool
+ //The number of transactionImpl coordinators
+ tcNum uint64
+ 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"
+
+/*
+*
+Init a transactionImpl coordinator client and acquire connections with all transactionImpl 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)
+ tc.cons = make([]internal.Connection, tc.tcNum)
+
+ //Get connections with all transaction_impl 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_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()
+ }
+}
+
+/*
+*
+New a transactionImpl 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
Review Comment:
```suggestion
return nil, err
```
##########
pulsar/transaction_coordinator_client.go:
##########
@@ -0,0 +1,223 @@
+// 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
+ blockIfReachMaxPendingOps bool
+ //The number of transactionImpl coordinators
+ tcNum uint64
+ 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"
+
+/*
+*
+Init a transactionImpl coordinator client and acquire connections with all transactionImpl 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)
+ tc.cons = make([]internal.Connection, tc.tcNum)
+
+ //Get connections with all transaction_impl 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_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()
+ }
+}
+
+/*
+*
+New a transactionImpl 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
Review Comment:
```suggestion
return nil, err
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] BewareMyPower commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1116433285
##########
integration-tests/conf/standalone.conf:
##########
@@ -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"}
Review Comment:
Why did you add the TLS authentication? Does transaction depend on the TLS?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] shibd commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1112724648
##########
pulsar/transactionCoordinatorClient.go:
##########
@@ -0,0 +1,222 @@
+// 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
+ blockIfReachMaxPendingOps bool
+ //The number of transactionImpl coordinators
+ tcNum uint64
+ 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"
+
+/*
+*
+Init a transactionImpl coordinator client and acquire connections with all transactionImpl 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)
+ tc.cons = make([]internal.Connection, tc.tcNum)
+
+ //Get connections with all transaction_impl 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_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()
+ }
+}
+
+/*
+*
+New a transactionImpl which can be used to guarantee exactly-once semantics.
+*/
+func (tc *transactionCoordinatorClient) newTransaction(timeout time.Duration) (TxnID, error) {
Review Comment:
This method will be changed to return `transaction` object in the future, right?
##########
Makefile:
##########
@@ -18,7 +18,7 @@
#
IMAGE_NAME = pulsar-client-go-test:latest
-PULSAR_VERSION ?= 2.8.3
+PULSAR_VERSION ?= 2.10.3
Review Comment:
Maybe we need to keep using `latest` version.
##########
pulsar/transaction_test.go:
##########
@@ -0,0 +1,100 @@
+// 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 := "my-topic"
Review Comment:
```suggestion
topic := newTopicName()
```
##########
pulsar/transaction_test.go:
##########
@@ -0,0 +1,100 @@
+// 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 := "my-topic"
+ 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()
+}
+
+/*
+Create a transaction coordinator client to send request
+*/
+func createTcClient(t *testing.T) (*transactionCoordinatorClient, *client) {
+ c, err := NewClient(ClientOptions{
+ URL: webServiceURLTLS,
Review Comment:
Does Transaction have to have TLS enabled?
##########
pulsar/transactionCoordinatorClient.go:
##########
@@ -0,0 +1,222 @@
+// 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
+ blockIfReachMaxPendingOps bool
+ //The number of transactionImpl coordinators
+ tcNum uint64
+ 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"
+
+/*
+*
+Init a transactionImpl coordinator client and acquire connections with all transactionImpl 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)
+ tc.cons = make([]internal.Connection, tc.tcNum)
+
+ //Get connections with all transaction_impl 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_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()
+ }
+}
+
+/*
+*
+New a transactionImpl 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
Review Comment:
We better use the `tcId` returned by the broker.
```suggestion
return TxnID{*cnx.Response.NewTxnResponse.TxnidMostBits, *cnx.Response.NewTxnResponse.TxnidLeastBits}, nil
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] liangyepianzhou commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1114069042
##########
pulsar/transaction_test.go:
##########
@@ -0,0 +1,100 @@
+// 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 := "my-topic"
+ 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()
+}
+
+/*
+Create a transaction coordinator client to send request
+*/
+func createTcClient(t *testing.T) (*transactionCoordinatorClient, *client) {
+ c, err := NewClient(ClientOptions{
+ URL: webServiceURLTLS,
Review Comment:
The transaction needs to get a connection from "pulsar/system/transaction_coordinator_assign".
And the client only has the authority of the `public` tenant in default.
So we need to add the authentication here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] liangyepianzhou commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1115215961
##########
pulsar/transaction_coordinator_client.go:
##########
@@ -0,0 +1,223 @@
+// 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
+ blockIfReachMaxPendingOps bool
+ //The number of transactionImpl coordinators
+ tcNum uint64
+ 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"
+
+/*
+*
+Init a transactionImpl coordinator client and acquire connections with all transactionImpl 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)
+ tc.cons = make([]internal.Connection, tc.tcNum)
+
+ //Get connections with all transaction_impl 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_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()
+ }
+}
+
+/*
+*
+New a transactionImpl 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{*cnx.Response.NewTxnResponse.TxnidMostBits,
+ *cnx.Response.NewTxnResponse.TxnidLeastBits}, nil
+}
+
+/*
+*
+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 {
+ 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 transactionImpl.
+And this can be used when ending the transactionImpl.
+*/
+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 transactionImpl.
+*/
+func (tc *transactionCoordinatorClient) endTxn(id *TxnID, action pb.TxnAction) error {
Review Comment:
This is the logic of internal calls, not the interface exposed to users. So it is not necessary to determine whether it is `nil` here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] liangyepianzhou commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1114057415
##########
Makefile:
##########
@@ -18,7 +18,7 @@
#
IMAGE_NAME = pulsar-client-go-test:latest
-PULSAR_VERSION ?= 2.8.3
+PULSAR_VERSION ?= 2.10.3
Review Comment:
CI failed again after I changed it to `latest`, so I can only change it to 2.10.3. The specific reason is not clear.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] nodece commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1114101702
##########
pulsar/transactionCoordinatorClient.go:
##########
@@ -0,0 +1,223 @@
+// 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
+ blockIfReachMaxPendingOps bool
+ //The number of transactionImpl coordinators
+ tcNum uint64
+ 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"
+
+/*
+*
+Init a transactionImpl coordinator client and acquire connections with all transactionImpl 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)
+ tc.cons = make([]internal.Connection, tc.tcNum)
+
+ //Get connections with all transaction_impl 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_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()
+ }
+}
+
+/*
+*
+New a transactionImpl which can be used to guarantee exactly-once semantics.
+*/
+func (tc *transactionCoordinatorClient) newTransaction(timeout time.Duration) (TxnID, error) {
Review Comment:
It is better `*TxnID, error`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] shibd commented on a diff in pull request #953: [feat][txn]Implement transactionCoordinatorClient
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #953:
URL: https://github.com/apache/pulsar-client-go/pull/953#discussion_r1114150219
##########
Makefile:
##########
@@ -18,7 +18,7 @@
#
IMAGE_NAME = pulsar-client-go-test:latest
-PULSAR_VERSION ?= 2.8.3
+PULSAR_VERSION ?= 2.10.3
Review Comment:
> CI failed again after I changed it to latest, so I can only change it to 2.10.3. The specific reason is not clear.
> Fixed version is right.
We can fix it in other PRs. But, Do we need the Go client to be compatible with the latest major version (the current is 2.11.0)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org