diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java index c2ce36d49ffe7..2a3213e10a210 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java @@ -73,6 +73,7 @@ import org.apache.pulsar.common.naming.NamespaceBundleSplitAlgorithm; import org.apache.pulsar.common.naming.NamespaceBundles; import org.apache.pulsar.common.naming.NamespaceName; +import org.apache.pulsar.common.naming.SystemTopicNames; import org.apache.pulsar.common.naming.TopicDomain; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.policies.data.AuthAction; @@ -306,14 +307,24 @@ protected void internalDeleteNamespace(AsyncResponse asyncResponse, boolean auth asyncResponse.resume(new RestException(e)); return; } - // remove from owned namespace map and ephemeral node from ZK final List> futures = Lists.newArrayList(); // remove system topics first. + Set noPartitionedTopicPolicySystemTopic = new HashSet<>(); + Set partitionedTopicPolicySystemTopic = new HashSet<>(); if (!topics.isEmpty()) { for (String topic : topics) { try { - futures.add(pulsar().getAdminClient().topics().deleteAsync(topic, true, true)); + if (SystemTopicNames.isTopicPoliciesSystemTopic(topic)) { + TopicName topicName = TopicName.get(topic); + if (topicName.isPartitioned()) { + partitionedTopicPolicySystemTopic.add(topic); + } else { + noPartitionedTopicPolicySystemTopic.add(topic); + } + } else { + futures.add(pulsar().getAdminClient().topics().deleteAsync(topic, true, true)); + } } catch (Exception ex) { log.error("[{}] Failed to delete system topic {}", clientAppId(), topic, ex); asyncResponse.resume(new RestException(Status.INTERNAL_SERVER_ERROR, ex)); @@ -321,11 +332,14 @@ protected void internalDeleteNamespace(AsyncResponse asyncResponse, boolean auth } } } - FutureUtil.waitForAll(futures).thenCompose(__ -> { - List> deleteBundleFutures = Lists.newArrayList(); - NamespaceBundles bundles = pulsar().getNamespaceService().getNamespaceBundleFactory() - .getBundles(namespaceName); - for (NamespaceBundle bundle : bundles.getBundles()) { + FutureUtil.waitForAll(futures) + .thenCompose(ignore -> internalDeleteTopicsAsync(noPartitionedTopicPolicySystemTopic)) + .thenCompose(ignore -> internalDeletePartitionedTopicsAsync(partitionedTopicPolicySystemTopic)) + .thenCompose(__ -> { + List> deleteBundleFutures = Lists.newArrayList(); + NamespaceBundles bundles = pulsar().getNamespaceService().getNamespaceBundleFactory() + .getBundles(namespaceName); + for (NamespaceBundle bundle : bundles.getBundles()) { // check if the bundle is owned by any broker, if not then we do not need to delete the bundle deleteBundleFutures.add(pulsar().getNamespaceService().getOwnerAsync(bundle).thenCompose(ownership -> { if (ownership.isPresent()) { @@ -475,27 +489,41 @@ protected void internalDeleteNamespaceForcefully(AsyncResponse asyncResponse, bo Set nonPartitionedTopics = new HashSet<>(); Set allSystemTopics = new HashSet<>(); Set allPartitionedSystemTopics = new HashSet<>(); + Set noPartitionedTopicPolicySystemTopic = new HashSet<>(); + Set partitionedTopicPolicySystemTopic = new HashSet<>(); for (String topic : topics) { try { TopicName topicName = TopicName.get(topic); if (topicName.isPartitioned()) { if (pulsar().getBrokerService().isSystemTopic(topicName)) { - allPartitionedSystemTopics.add(topicName.getPartitionedTopicName()); + if (SystemTopicNames.isTopicPoliciesSystemTopic(topic)) { + partitionedTopicPolicySystemTopic.add(topicName.getPartitionedTopicName()); + } else { + allPartitionedSystemTopics.add(topicName.getPartitionedTopicName()); + } continue; } String partitionedTopic = topicName.getPartitionedTopicName(); if (!partitionedTopics.contains(partitionedTopic)) { + // Distinguish partitioned topic to avoid duplicate deletion of the same schema + topicFutures.add(pulsar().getAdminClient().topics().deletePartitionedTopicAsync( + partitionedTopic, true, true)); partitionedTopics.add(partitionedTopic); } } else { if (pulsar().getBrokerService().isSystemTopic(topicName)) { - allSystemTopics.add(topic); + if (SystemTopicNames.isTopicPoliciesSystemTopic(topic)) { + noPartitionedTopicPolicySystemTopic.add(topic); + } else { + allSystemTopics.add(topic); + } continue; } + topicFutures.add(pulsar().getAdminClient().topics().deleteAsync( + topic, true, true)); nonPartitionedTopics.add(topic); } - topicFutures.add(pulsar().getAdminClient().topics().deleteAsync(topic, true)); } catch (Exception e) { String errorMessage = String.format("Failed to force delete topic %s, " + "but the previous deletion command of partitioned-topics:%s " @@ -508,11 +536,6 @@ protected void internalDeleteNamespaceForcefully(AsyncResponse asyncResponse, bo } } - for (String partitionedTopic : partitionedTopics) { - topicFutures.add(namespaceResources().getPartitionedTopicResources() - .deletePartitionedTopicAsync(TopicName.get(partitionedTopic))); - } - if (log.isDebugEnabled()) { log.debug("Successfully send deletion command of partitioned-topics:{} " + "and non-partitioned-topics:{} in namespace:{}.", @@ -524,6 +547,9 @@ protected void internalDeleteNamespaceForcefully(AsyncResponse asyncResponse, bo .thenCompose((ignore) -> internalDeleteTopicsAsync(allSystemTopics)) .thenCompose((ignore) -> internalDeletePartitionedTopicsAsync(allPartitionedSystemTopics)) + .thenCompose(ignore -> + internalDeletePartitionedTopicsAsync(partitionedTopicPolicySystemTopic)) + .thenCompose(ignore -> internalDeleteTopicsAsync(noPartitionedTopicPolicySystemTopic)) .handle((result, exception) -> { if (exception != null) { if (exception.getCause() instanceof PulsarAdminException) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index 8e95f2c431361..61e6c6b1a0635 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -1183,10 +1183,10 @@ private CompletableFuture delete(boolean failIfHasSubscriptions, closeClientFuture.thenAccept(__ -> { CompletableFuture deleteTopicAuthenticationFuture = new CompletableFuture<>(); brokerService.deleteTopicAuthenticationWithRetry(topic, deleteTopicAuthenticationFuture, 5); - deleteTopicAuthenticationFuture.thenCompose(ignore -> deleteSchema()) + + deleteTopicAuthenticationFuture.thenCompose(ignore -> deleteSchema()) .thenCompose(ignore -> { - if (!SystemTopicNames.isTopicPoliciesSystemTopic(topic) - && brokerService.getPulsar().getConfiguration().isSystemTopicEnabled()) { + if (!SystemTopicNames.isTopicPoliciesSystemTopic(topic)) { return deleteTopicPolicies(); } else { return CompletableFuture.completedFuture(null); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java index 16dfe5bc9a3b6..eaa54c1fd3402 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.broker.admin; +import static org.apache.pulsar.common.naming.NamespaceName.SYSTEM_NAMESPACE; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doNothing; import static org.mockito.Mockito.doReturn; @@ -57,11 +58,13 @@ import javax.ws.rs.core.Response.Status; import javax.ws.rs.core.UriBuilder; import javax.ws.rs.core.UriInfo; +import lombok.Cleanup; import org.apache.bookkeeper.client.api.ReadHandle; import org.apache.bookkeeper.mledger.LedgerOffloader; import org.apache.bookkeeper.mledger.ManagedLedgerConfig; import org.apache.bookkeeper.util.ZkUtils; import org.apache.pulsar.broker.BrokerTestUtil; +import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.admin.v1.Namespaces; import org.apache.pulsar.broker.admin.v1.PersistentTopics; import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; @@ -81,6 +84,7 @@ import org.apache.pulsar.client.api.ConsumerBuilder; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.common.api.proto.CommandSubscribe; import org.apache.pulsar.common.naming.NamespaceBundle; @@ -103,6 +107,7 @@ import org.apache.pulsar.common.policies.data.SubscribeRate; import org.apache.pulsar.common.policies.data.TenantInfo; import org.apache.pulsar.common.policies.data.TenantInfoImpl; +import org.apache.pulsar.common.policies.data.TopicPolicies; import org.apache.pulsar.common.policies.data.impl.DispatchRateImpl; import org.apache.pulsar.metadata.cache.impl.MetadataCacheImpl; import org.apache.pulsar.metadata.impl.AbstractMetadataStore; @@ -1938,7 +1943,7 @@ public void testFinallyDeleteSystemTopicWhenDeleteNamespace() throws Exception { } @Test - public void testNotClearTopicPolicesWhenDeleteSystemTopic() throws Exception { + public void testNotClearTopicPolicesWhenDeleteTopicPolicyTopic() throws Exception { String namespace = this.testTenant + "/delete-systemTopic"; String topic = TopicName.get(TopicDomain.persistent.toString(), this.testTenant, "delete-systemTopic", "testNotClearTopicPolicesWhenDeleteSystemTopic").toString(); @@ -1958,4 +1963,33 @@ public void testNotClearTopicPolicesWhenDeleteSystemTopic() throws Exception { // 4. delete the policies topic and the topic wil not to clear topic polices admin.topics().delete(namespace + "/" + SystemTopicNames.NAMESPACE_EVENTS_LOCAL_NAME, true); } + @Test + public void testDeleteTopicPolicyWhenDeleteSystemTopic() throws Exception { + conf.setTopicLevelPoliciesEnabled(true); + conf.setSystemTopicEnabled(true); + Field field = PulsarService.class.getDeclaredField("topicPoliciesService"); + field.setAccessible(true); + field.set(pulsar, new SystemTopicBasedTopicPoliciesService(pulsar)); + + String systemTopic = SYSTEM_NAMESPACE.toString() + "/" + "testDeleteTopicPolicyWhenDeleteSystemTopic"; + admin.tenants().createTenant(SYSTEM_NAMESPACE.getTenant(), + new TenantInfoImpl(Set.of("role1", "role2"), Set.of("use", "usc", "usw"))); + + admin.namespaces().createNamespace(SYSTEM_NAMESPACE.toString()); + @Cleanup + Producer producer = pulsarClient.newProducer(Schema.STRING) + .topic(systemTopic).create(); + admin.topicPolicies().setMaxConsumers(systemTopic, 5); + + Integer maxConsumerPerTopic = pulsar + .getTopicPoliciesService() + .getTopicPoliciesBypassCacheAsync(TopicName.get(systemTopic)).get() + .getMaxConsumerPerTopic(); + + assertEquals(maxConsumerPerTopic, Integer.valueOf(5)); + admin.topics().delete(systemTopic, true); + TopicPolicies topicPolicies = pulsar.getTopicPoliciesService() + .getTopicPoliciesBypassCacheAsync(TopicName.get(systemTopic)).get(5, TimeUnit.SECONDS); + assertNull(topicPolicies); + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java index ccdca13c996e9..dc3b4d6f0f26d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java @@ -1472,6 +1472,54 @@ public Object answer(InvocationOnMock invocation) throws Throwable { } } + @Test + public void testGetTxnState() throws Exception { + Transaction transaction = pulsarClient.newTransaction().withTransactionTimeout(1, TimeUnit.SECONDS) + .build().get(); + + // test OPEN and TIMEOUT + assertEquals(transaction.getState(), Transaction.State.OPEN); + Transaction timeoutTxn = transaction; + Awaitility.await().until(() -> timeoutTxn.getState() == Transaction.State.TIME_OUT); + + // test abort + transaction = pulsarClient.newTransaction().withTransactionTimeout(3, TimeUnit.SECONDS) + .build().get(); + transaction.abort().get(); + assertEquals(transaction.getState(), Transaction.State.ABORTED); + + // test commit + transaction = pulsarClient.newTransaction().withTransactionTimeout(3, TimeUnit.SECONDS) + .build().get(); + transaction.commit().get(); + assertEquals(transaction.getState(), Transaction.State.COMMITTED); + + // test error + transaction = pulsarClient.newTransaction().withTransactionTimeout(1, TimeUnit.SECONDS) + .build().get(); + pulsarServiceList.get(0).getTransactionMetadataStoreService() + .endTransaction(transaction.getTxnID(), 0, false); + transaction.commit(); + Transaction errorTxn = transaction; + Awaitility.await().until(() -> errorTxn.getState() == Transaction.State.ERROR); + + // test committing + transaction = pulsarClient.newTransaction().withTransactionTimeout(3, TimeUnit.SECONDS) + .build().get(); + ((TransactionImpl) transaction).registerSendOp(new CompletableFuture<>()); + transaction.commit(); + Transaction committingTxn = transaction; + Awaitility.await().until(() -> committingTxn.getState() == Transaction.State.COMMITTING); + + // test aborting + transaction = pulsarClient.newTransaction().withTransactionTimeout(3, TimeUnit.SECONDS) + .build().get(); + ((TransactionImpl) transaction).registerSendOp(new CompletableFuture<>()); + transaction.abort(); + Transaction abortingTxn = transaction; + Awaitility.await().until(() -> abortingTxn.getState() == Transaction.State.ABORTING); + } + @Test public void testEncryptionRequired() throws Exception { final String namespace = "tnx/ns-prechecks"; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientDeduplicationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientDeduplicationTest.java index c8acc7d46f82d..0ac0440a7a8b2 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientDeduplicationTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientDeduplicationTest.java @@ -28,6 +28,9 @@ import java.util.Arrays; import java.util.List; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import lombok.extern.slf4j.Slf4j; @@ -373,4 +376,51 @@ public void testKeyBasedBatchingOrder() throws Exception { consumer.close(); producer.close(); } + + @Test + public void testUpdateSequenceIdInSyncCodeSegment() throws Exception { + final String topic = "persistent://my-property/my-ns/testUpdateSequenceIdInSyncCodeSegment"; + int totalMessage = 200; + int threadSize = 5; + String topicName = "subscription"; + ExecutorService executorService = Executors.newFixedThreadPool(threadSize); + conf.setBrokerDeduplicationEnabled(true); + + //build producer/consumer + Producer producer = pulsarClient.newProducer() + .topic(topic) + .producerName("producer") + .sendTimeout(0, TimeUnit.SECONDS) + .create(); + + Consumer consumer = pulsarClient.newConsumer() + .topic(topic) + .subscriptionType(SubscriptionType.Exclusive) + .subscriptionName(topicName) + .subscribe(); + + CountDownLatch countDownLatch = new CountDownLatch(threadSize); + //Send messages in multiple-thread + for (int i = 0; i < threadSize; i++) { + executorService.submit(() -> { + try { + for (int j = 0; j < totalMessage; j++) { + //The message will be sent with out-of-order sequence ID. + producer.newMessage().sendAsync(); + } + } catch (Exception e) { + log.error("Failed to send/ack messages with transaction.", e); + } finally { + countDownLatch.countDown(); + } + }); + } + //wait the all send op is executed and store its futures in the arraylist. + countDownLatch.await(); + + for (int i = 0; i < threadSize * totalMessage; i++) { + Message msg = consumer.receive(5, TimeUnit.SECONDS); + assertNotNull(msg); + } + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java index bc56eab6bc15d..454edebe6f08d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java @@ -1186,7 +1186,7 @@ public void testTxnTimeOutInClient() throws Exception{ .build().get(); producer.newMessage().send(); Awaitility.await().untilAsserted(() -> { - Assert.assertEquals(((TransactionImpl)transaction).getState(), TransactionImpl.State.TIMEOUT); + Assert.assertEquals(((TransactionImpl)transaction).getState(), TransactionImpl.State.TIME_OUT); }); try { diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/transaction/Transaction.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/transaction/Transaction.java index fd4cf0bc1665c..33e96d5c2764d 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/transaction/Transaction.java +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/transaction/Transaction.java @@ -29,6 +29,55 @@ @InterfaceStability.Evolving public interface Transaction { + enum State { + + /** + * When a transaction is in the `OPEN` state, messages can be produced and acked with this transaction. + * + * When a transaction is in the `OPEN` state, it can commit or abort. + */ + OPEN, + + /** + * When a client invokes a commit, the transaction state is changed from `OPEN` to `COMMITTING`. + */ + COMMITTING, + + /** + * When a client invokes an abort, the transaction state is changed from `OPEN` to `ABORTING`. + */ + ABORTING, + + /** + * When a client receives a response to a commit, the transaction state is changed from + * `COMMITTING` to `COMMITTED`. + */ + COMMITTED, + + /** + * When a client receives a response to an abort, the transaction state is changed from `ABORTING` to `ABORTED`. + */ + ABORTED, + + /** + * When a client invokes a commit or an abort, but a transaction does not exist in a coordinator, + * then the state is changed to `ERROR`. + * + * When a client invokes a commit, but the transaction state in a coordinator is `ABORTED` or `ABORTING`, + * then the state is changed to `ERROR`. + * + * When a client invokes an abort, but the transaction state in a coordinator is `COMMITTED` or `COMMITTING`, + * then the state is changed to `ERROR`. + */ + ERROR, + + /** + * When a transaction is timed out and the transaction state is `OPEN`, + * then the transaction state is changed from `OPEN` to `TIME_OUT`. + */ + TIME_OUT + } + /** * Commit the transaction. * @@ -48,4 +97,12 @@ public interface Transaction { * @return {@link TxnID} the txnID. */ TxnID getTxnID(); + + /** + * Get transaction state. + * + * @return {@link State} the state of the transaction. + */ + State getState(); + } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java index 6f7d7e6a148fe..3969d1b2afe88 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java @@ -101,7 +101,7 @@ public class ProducerImpl extends ProducerBase implements TimerTask, Conne // Producer id, used to identify a producer within a single connection protected final long producerId; - // Variable is used through the atomic updater + // Variable is updated in a synchronized block private volatile long msgIdGenerator; private final OpSendMsgQueue pendingMessages; @@ -169,10 +169,6 @@ public class ProducerImpl extends ProducerBase implements TimerTask, Conne private boolean errorState; - @SuppressWarnings("rawtypes") - private static final AtomicLongFieldUpdater msgIdGeneratorUpdater = AtomicLongFieldUpdater - .newUpdater(ProducerImpl.class, "msgIdGenerator"); - public ProducerImpl(PulsarClientImpl client, String topic, ProducerConfigurationData conf, CompletableFuture> producerCreatedFuture, int partitionIndex, Schema schema, ProducerInterceptors interceptors, Optional overrideProducerName) { @@ -489,7 +485,7 @@ public void sendAsync(Message message, SendCallback callback) { // Update the message metadata before computing the payload chunk size to avoid a large message cannot be split // into chunks. - final long sequenceId = updateMessageMetadata(msgMetadata, uncompressedSize); + updateMessageMetadata(msgMetadata, uncompressedSize); // send in chunks int totalChunks; @@ -529,6 +525,7 @@ public void sendAsync(Message message, SendCallback callback) { try { synchronized (this) { int readStartIndex = 0; + final long sequenceId = updateMessageMetadataSequenceId(msgMetadata); String uuid = totalChunks > 1 ? String.format("%s-%d", producerName, sequenceId) : null; ChunkedMessageCtx chunkedMessageCtx = totalChunks > 1 ? ChunkedMessageCtx.get(totalChunks) : null; byte[] schemaVersion = totalChunks > 1 && msg.getMessageBuilder().hasSchemaVersion() @@ -570,15 +567,7 @@ public void sendAsync(Message message, SendCallback callback) { * @param uncompressedSize * @return the sequence id */ - private long updateMessageMetadata(final MessageMetadata msgMetadata, final int uncompressedSize) { - final long sequenceId; - if (!msgMetadata.hasSequenceId()) { - sequenceId = msgIdGeneratorUpdater.getAndIncrement(this); - msgMetadata.setSequenceId(sequenceId); - } else { - sequenceId = msgMetadata.getSequenceId(); - } - + private void updateMessageMetadata(final MessageMetadata msgMetadata, final int uncompressedSize) { if (!msgMetadata.hasPublishTime()) { msgMetadata.setPublishTime(client.getClientClock().millis()); @@ -592,6 +581,16 @@ private long updateMessageMetadata(final MessageMetadata msgMetadata, final int } msgMetadata.setUncompressedSize(uncompressedSize); } + } + + private long updateMessageMetadataSequenceId(final MessageMetadata msgMetadata) { + final long sequenceId; + if (!msgMetadata.hasSequenceId()) { + sequenceId = msgIdGenerator++; + msgMetadata.setSequenceId(sequenceId); + } else { + sequenceId = msgMetadata.getSequenceId(); + } return sequenceId; } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/TransactionImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/TransactionImpl.java index 55b20438693e3..833b0957d1c8a 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/TransactionImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/TransactionImpl.java @@ -70,17 +70,7 @@ public class TransactionImpl implements Transaction , TimerTask { @Override public void run(Timeout timeout) throws Exception { - STATE_UPDATE.compareAndSet(this, State.OPEN, State.TIMEOUT); - } - - public enum State { - OPEN, - COMMITTING, - ABORTING, - COMMITTED, - ABORTED, - ERROR, - TIMEOUT + STATE_UPDATE.compareAndSet(this, State.OPEN, State.TIME_OUT); } TransactionImpl(PulsarClientImpl client, @@ -215,6 +205,11 @@ public TxnID getTxnID() { return new TxnID(txnIdMostBits, txnIdLeastBits); } + @Override + public State getState() { + return state; + } + public boolean checkIfOpen(CompletableFuture completableFuture) { if (state == State.OPEN) { return true;