-
Notifications
You must be signed in to change notification settings - Fork 3.6k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Add getTxnID method in Transaction.java #11438
Changes from 3 commits
f6ecfd7
f5c5499
8fd8e65
1b377cd
ee4daff
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -19,6 +19,8 @@ | |
package org.apache.pulsar.broker.transaction; | ||
|
||
import static org.apache.pulsar.transaction.coordinator.impl.MLTransactionLogImpl.TRANSACTION_LOG_PREFIX; | ||
import static org.testng.AssertJUnit.assertEquals; | ||
import static org.testng.AssertJUnit.assertNotNull; | ||
import com.google.common.collect.Sets; | ||
import java.util.Optional; | ||
import java.util.concurrent.CompletableFuture; | ||
|
@@ -32,12 +34,18 @@ | |
import org.apache.pulsar.client.api.PulsarClient; | ||
import org.apache.pulsar.client.api.PulsarClientException; | ||
import org.apache.pulsar.client.api.SubscriptionType; | ||
import org.apache.pulsar.client.api.transaction.Transaction; | ||
import org.apache.pulsar.client.api.transaction.TxnID; | ||
import org.apache.pulsar.common.naming.NamespaceName; | ||
import org.apache.pulsar.common.naming.TopicDomain; | ||
import org.apache.pulsar.common.naming.TopicName; | ||
import org.apache.pulsar.common.policies.data.ClusterData; | ||
import org.apache.pulsar.common.policies.data.TenantInfoImpl; | ||
import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap; | ||
import org.apache.pulsar.transaction.coordinator.TransactionCoordinatorID; | ||
import org.apache.pulsar.transaction.coordinator.TransactionMetadataStore; | ||
import org.apache.pulsar.transaction.coordinator.TransactionMetadataStoreState; | ||
import org.apache.pulsar.transaction.coordinator.impl.MLTransactionMetadataStore; | ||
import org.awaitility.Awaitility; | ||
import org.testng.Assert; | ||
import org.testng.annotations.AfterMethod; | ||
|
@@ -53,23 +61,26 @@ public class TransactionTest extends TransactionTestBase { | |
|
||
private static final String TENANT = "tnx"; | ||
private static final String NAMESPACE1 = TENANT + "/ns1"; | ||
private static final int NUM_BROKER = 1; | ||
private static final int NUM_TC_PER_ = 1; | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. is there a typo here ? why this variable ends with a '_' ? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. terribly sorry. |
||
|
||
@BeforeMethod | ||
protected void setup() throws Exception { | ||
this.setBrokerCount(1); | ||
this.setBrokerCount(NUM_BROKER); | ||
this.internalSetup(); | ||
|
||
String[] brokerServiceUrlArr = getPulsarServiceList().get(0).getBrokerServiceUrl().split(":"); | ||
String[] brokerServiceUrlArr = getPulsarServiceList().get(NUM_BROKER - 1).getBrokerServiceUrl().split(":"); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. why we are picking the last and not the first ? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Excuse me. |
||
String webServicePort = brokerServiceUrlArr[brokerServiceUrlArr.length - 1]; | ||
admin.clusters().createCluster(CLUSTER_NAME, ClusterData.builder().serviceUrl("http://localhost:" + webServicePort).build()); | ||
admin.clusters().createCluster(CLUSTER_NAME, ClusterData.builder() | ||
.serviceUrl("http://localhost:" + webServicePort).build()); | ||
admin.tenants().createTenant(TENANT, | ||
new TenantInfoImpl(Sets.newHashSet("appid1"), Sets.newHashSet(CLUSTER_NAME))); | ||
admin.namespaces().createNamespace(NAMESPACE1); | ||
|
||
admin.tenants().createTenant(NamespaceName.SYSTEM_NAMESPACE.getTenant(), | ||
new TenantInfoImpl(Sets.newHashSet("appid1"), Sets.newHashSet(CLUSTER_NAME))); | ||
admin.namespaces().createNamespace(NamespaceName.SYSTEM_NAMESPACE.toString()); | ||
admin.topics().createPartitionedTopic(TopicName.TRANSACTION_COORDINATOR_ASSIGN.toString(), 1); | ||
admin.topics().createPartitionedTopic(TopicName.TRANSACTION_COORDINATOR_ASSIGN.toString(), NUM_TC_PER_); | ||
pulsarClient.close(); | ||
pulsarClient = PulsarClient.builder() | ||
.serviceUrl(getPulsarServiceList().get(0).getBrokerServiceUrl()) | ||
|
@@ -139,4 +150,21 @@ public Consumer<byte[]> getConsumer(String topicName, String subName) throws Pul | |
.enableBatchIndexAcknowledgment(true) | ||
.subscribe(); | ||
} | ||
|
||
@Test | ||
public void testGetTxnID() throws Exception { | ||
// wait tc init success to ready state | ||
Assert.assertTrue(waitForCoordinatorToBeAvailable(NUM_BROKER, NUM_TC_PER_)); | ||
Transaction transaction = pulsarClient.newTransaction() | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Why need to create the transaction again? |
||
.build().get(); | ||
TxnID txnID = transaction.getTxnID(); | ||
Assert.assertEquals(txnID.getLeastSigBits(), 0); | ||
Assert.assertEquals(txnID.getMostSigBits(), 0); | ||
transaction.abort(); | ||
transaction = pulsarClient.newTransaction() | ||
.build().get(); | ||
txnID = transaction.getTxnID(); | ||
Assert.assertEquals(txnID.getLeastSigBits(), 1); | ||
Assert.assertEquals(txnID.getMostSigBits(), 0); | ||
Comment on lines
+167
to
+168
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We will always get transaction ID (1,0) here? I mean we can just check the txnID is not null. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. In fact it should always be this |
||
} | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
what about NUM_BROKERS ?