You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2021/03/02 13:12:20 UTC
[GitHub] [pulsar] congbobo184 opened a new pull request #9776: Congbobo184 transaction client transaction state check
congbobo184 opened a new pull request #9776:
URL: https://github.com/apache/pulsar/pull/9776
## Motivation
Now pulsar client don't check the transaction state, so when it committed or aborted, it also can produce or ack message. It should be prevented.
## implement
Add the State :
```
public enum State {
OPEN,
CLOSE
}
```
when open we can do transaction operation, it is close we can't do transaction operation.
### Verifying this change
Add the tests for it
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): (no)
The public API: (no)
The schema: (no)
The default values of configurations: (no)
The wire protocol: (yes)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #9776: [Pulsar Transaction]Transaction client transaction state check
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #9776:
URL: https://github.com/apache/pulsar/pull/9776#discussion_r585996155
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/TransactionImpl.java
##########
@@ -63,11 +61,18 @@
private final ArrayList<CompletableFuture<MessageId>> sendFutureList;
private final ArrayList<CompletableFuture<Void>> ackFutureList;
+ private volatile State state;
+
+ public enum State {
+ OPEN,
+ CLOSE
Review comment:
Why not keep the state consistent with the transaction state? Open/Committing/Aborting/Committed/Aborted ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #9776: [Pulsar Transaction]Transaction client transaction state check
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #9776:
URL: https://github.com/apache/pulsar/pull/9776#issuecomment-792708228
@gaoran10 Please help review this PR
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on pull request #9776: [Pulsar Transaction]Transaction client transaction state check
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on pull request #9776:
URL: https://github.com/apache/pulsar/pull/9776#issuecomment-792403657
/pulsarbot run-failure-checks
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #9776: [Pulsar Transaction]Transaction client transaction state check
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #9776:
URL: https://github.com/apache/pulsar/pull/9776#discussion_r589366260
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/TransactionImpl.java
##########
@@ -126,52 +143,76 @@ public synchronized void registerCumulativeAckConsumer(ConsumerImpl<?> consumer)
@Override
public CompletableFuture<Void> commit() {
- CompletableFuture<Void> commitFuture = new CompletableFuture<>();
- allOpComplete().whenComplete((v, e) -> {
- if (e != null) {
- abort().whenComplete((vx, ex) -> commitFuture.completeExceptionally(e));
- } else {
- tcClient.commitAsync(new TxnID(txnIdMostBits, txnIdLeastBits))
- .whenComplete((vx, ex) -> {
- if (ex != null) {
- commitFuture.completeExceptionally(ex);
- } else {
- commitFuture.complete(vx);
- }
- });
- }
+ return checkIfOpen().thenCompose((value) -> {
+ CompletableFuture<Void> commitFuture = new CompletableFuture<>();
+ this.state = State.COMMITTING;
+ allOpComplete().whenComplete((v, e) -> {
+ if (e != null) {
+ abort().whenComplete((vx, ex) -> commitFuture.completeExceptionally(e));
+ } else {
+ tcClient.commitAsync(new TxnID(txnIdMostBits, txnIdLeastBits))
+ .whenComplete((vx, ex) -> {
+ if (ex != null) {
+ if (ex instanceof TransactionNotFoundException
+ || ex instanceof InvalidTxnStatusException) {
+ this.state = State.ERROR;
+ }
+ commitFuture.completeExceptionally(ex);
+ } else {
+ this.state = State.COMMITTED;
+ commitFuture.complete(vx);
+ }
+ });
+ }
+ });
+ return commitFuture;
});
- return commitFuture;
}
@Override
public CompletableFuture<Void> abort() {
- CompletableFuture<Void> abortFuture = new CompletableFuture<>();
- allOpComplete().whenComplete((v, e) -> {
- if (e != null) {
- log.error(e.getMessage());
- }
- if (cumulativeAckConsumers != null) {
- cumulativeAckConsumers.forEach((consumer, integer) ->
- cumulativeAckConsumers
- .putIfAbsent(consumer, consumer.clearIncomingMessagesAndGetMessageNumber()));
- }
- tcClient.abortAsync(new TxnID(txnIdMostBits, txnIdLeastBits)).whenComplete((vx, ex) -> {
+ return checkIfOpen().thenCompose(value -> {
+ CompletableFuture<Void> abortFuture = new CompletableFuture<>();
Review comment:
Change to aborting first?
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
##########
@@ -676,4 +685,82 @@ public void produceTxnMessageOrderTest() throws Exception {
}
}
+ @Test
+ public void produceAndConsumeCloseStateTxnTest() throws Exception {
+ String topic = NAMESPACE1 + "/txn-close-state";
+
+ @Cleanup
+ Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionName("test")
+ .subscribe();
+
+ @Cleanup
+ Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(topic)
+ .sendTimeout(0, TimeUnit.SECONDS)
+ .producerName("txn-close-state")
+ .create();
+
+ Transaction produceTxn = pulsarClient
+ .newTransaction()
+ .withTransactionTimeout(2, TimeUnit.SECONDS)
+ .build().get();
+
+ Transaction consumeTxn = pulsarClient
+ .newTransaction()
+ .withTransactionTimeout(2, TimeUnit.SECONDS)
+ .build().get();
+
+ producer.newMessage(produceTxn).value(("Hello Pulsar!").getBytes()).sendAsync().get();
+ produceTxn.commit().get();
+ try {
+ producer.newMessage(produceTxn).value(("Hello Pulsar!").getBytes()).sendAsync().get();
+ fail();
+ } catch (Exception e) {
+ assertTrue(e.getCause() instanceof TransactionCoordinatorClientException.InvalidTxnStatusException);
+ }
+
+ try {
+ produceTxn.commit().get();
+ fail();
+ } catch (Exception e) {
+ assertTrue(e.getCause() instanceof TransactionCoordinatorClientException.InvalidTxnStatusException);
+ }
+
+
+ Message<byte[]> message = consumer.receive(5, TimeUnit.SECONDS);
+ consumer.acknowledgeAsync(message.getMessageId(), consumeTxn).get();
+ consumeTxn.commit().get();
+ try {
+ consumer.acknowledgeAsync(message.getMessageId(), consumeTxn).get();
+ fail();
+ } catch (Exception e) {
+ assertTrue(e.getCause() instanceof TransactionCoordinatorClientException.InvalidTxnStatusException);
+ }
+
+ try {
+ consumeTxn.commit().get();
+ fail();
+ } catch (Exception e) {
+ assertTrue(e.getCause() instanceof TransactionCoordinatorClientException.InvalidTxnStatusException);
+ }
+
+ Transaction timeoutTxn = pulsarClient
+ .newTransaction()
+ .withTransactionTimeout(1, TimeUnit.SECONDS)
+ .build().get();
+
+ Thread.sleep(2000);
Review comment:
Please avoid use sleep.
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/TransactionImpl.java
##########
@@ -126,52 +143,76 @@ public synchronized void registerCumulativeAckConsumer(ConsumerImpl<?> consumer)
@Override
public CompletableFuture<Void> commit() {
- CompletableFuture<Void> commitFuture = new CompletableFuture<>();
- allOpComplete().whenComplete((v, e) -> {
- if (e != null) {
- abort().whenComplete((vx, ex) -> commitFuture.completeExceptionally(e));
- } else {
- tcClient.commitAsync(new TxnID(txnIdMostBits, txnIdLeastBits))
- .whenComplete((vx, ex) -> {
- if (ex != null) {
- commitFuture.completeExceptionally(ex);
- } else {
- commitFuture.complete(vx);
- }
- });
- }
+ return checkIfOpen().thenCompose((value) -> {
+ CompletableFuture<Void> commitFuture = new CompletableFuture<>();
+ this.state = State.COMMITTING;
+ allOpComplete().whenComplete((v, e) -> {
+ if (e != null) {
+ abort().whenComplete((vx, ex) -> commitFuture.completeExceptionally(e));
+ } else {
+ tcClient.commitAsync(new TxnID(txnIdMostBits, txnIdLeastBits))
+ .whenComplete((vx, ex) -> {
+ if (ex != null) {
+ if (ex instanceof TransactionNotFoundException
+ || ex instanceof InvalidTxnStatusException) {
+ this.state = State.ERROR;
+ }
+ commitFuture.completeExceptionally(ex);
+ } else {
+ this.state = State.COMMITTED;
+ commitFuture.complete(vx);
+ }
+ });
+ }
+ });
+ return commitFuture;
});
- return commitFuture;
}
@Override
public CompletableFuture<Void> abort() {
- CompletableFuture<Void> abortFuture = new CompletableFuture<>();
- allOpComplete().whenComplete((v, e) -> {
- if (e != null) {
- log.error(e.getMessage());
- }
- if (cumulativeAckConsumers != null) {
- cumulativeAckConsumers.forEach((consumer, integer) ->
- cumulativeAckConsumers
- .putIfAbsent(consumer, consumer.clearIncomingMessagesAndGetMessageNumber()));
- }
- tcClient.abortAsync(new TxnID(txnIdMostBits, txnIdLeastBits)).whenComplete((vx, ex) -> {
+ return checkIfOpen().thenCompose(value -> {
+ CompletableFuture<Void> abortFuture = new CompletableFuture<>();
+ allOpComplete().whenComplete((v, e) -> {
+ if (e != null) {
+ log.error(e.getMessage());
+ }
if (cumulativeAckConsumers != null) {
- cumulativeAckConsumers.forEach(ConsumerImpl::increaseAvailablePermits);
- cumulativeAckConsumers.clear();
+ cumulativeAckConsumers.forEach((consumer, integer) ->
+ cumulativeAckConsumers
+ .putIfAbsent(consumer, consumer.clearIncomingMessagesAndGetMessageNumber()));
}
+ tcClient.abortAsync(new TxnID(txnIdMostBits, txnIdLeastBits)).whenComplete((vx, ex) -> {
+ if (cumulativeAckConsumers != null) {
+ cumulativeAckConsumers.forEach(ConsumerImpl::increaseAvailablePermits);
+ cumulativeAckConsumers.clear();
+ }
- if (ex != null) {
- abortFuture.completeExceptionally(ex);
- } else {
- abortFuture.complete(null);
- }
+ if (ex != null) {
+ if (ex instanceof TransactionNotFoundException
+ || ex instanceof InvalidTxnStatusException) {
+ this.state = State.ERROR;
+ }
+ abortFuture.completeExceptionally(ex);
+ } else {
+ this.state = State.ABORTED;
+ abortFuture.complete(null);
+ }
+ });
});
+
+ return abortFuture;
});
+ }
- return abortFuture;
+ private CompletableFuture<Void> checkIfOpen() {
+ if (state == State.OPEN) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return FutureUtil.failedFuture(new InvalidTxnStatusException("[" + txnIdMostBits + ":"
+ + txnIdLeastBits + "] Transaction in " + state.name() + " state!"));
Review comment:
print the expected transaction state and the current transaction state?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #9776: [Pulsar Transaction]Transaction client transaction state check
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #9776:
URL: https://github.com/apache/pulsar/pull/9776#discussion_r586162476
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/TransactionImpl.java
##########
@@ -63,11 +61,18 @@
private final ArrayList<CompletableFuture<MessageId>> sendFutureList;
private final ArrayList<CompletableFuture<Void>> ackFutureList;
+ private volatile State state;
+
+ public enum State {
+ OPEN,
+ CLOSE
Review comment:
some situation we can't get the real state fro this transaction, example: the transaction timeout, it will be aborted but client commit.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on pull request #9776: [Pulsar Transaction]Transaction client transaction state check
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on pull request #9776:
URL: https://github.com/apache/pulsar/pull/9776#issuecomment-793270824
/pulsarbot run-failure-checks
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui merged pull request #9776: [Pulsar Transaction]Transaction client transaction state check
Posted by GitBox <gi...@apache.org>.
codelipenghui merged pull request #9776:
URL: https://github.com/apache/pulsar/pull/9776
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on pull request #9776: [Pulsar Transaction]Transaction client transaction state check
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on pull request #9776:
URL: https://github.com/apache/pulsar/pull/9776#issuecomment-792471168
/pulsarbot run-failure-checks
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #9776: [Pulsar Transaction]Transaction client transaction state check
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #9776:
URL: https://github.com/apache/pulsar/pull/9776#discussion_r586188447
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/TransactionImpl.java
##########
@@ -63,11 +61,18 @@
private final ArrayList<CompletableFuture<MessageId>> sendFutureList;
private final ArrayList<CompletableFuture<Void>> ackFutureList;
+ private volatile State state;
+
+ public enum State {
+ OPEN,
+ CLOSE
Review comment:
commit or abort finish, the state will become closed. we don't need to change the state to committing or aborting, it seem not useful. When we need committing or aborting sate we can add.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #9776: [Pulsar Transaction]Transaction client transaction state check
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #9776:
URL: https://github.com/apache/pulsar/pull/9776#discussion_r586176706
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/TransactionImpl.java
##########
@@ -63,11 +61,18 @@
private final ArrayList<CompletableFuture<MessageId>> sendFutureList;
private final ArrayList<CompletableFuture<Void>> ackFutureList;
+ private volatile State state;
+
+ public enum State {
+ OPEN,
+ CLOSE
Review comment:
So the state should be committing -> closed ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org