[Transaction] Define how the transaction client API is used

Authors: @congbobo184 @codelipenghui
# Background
Now more and more people and companies are testing and using transactions. However, the definition of transaction API is not clear. Users don't know how to try to catch Exceptions and what time do the transactions can commit or abort.
# Motivation
Let's take a look at the transaction current use of API.
```
Message message = consumer1.receive();
try {
Transaction transaction = client.newTransaction().withTransactionTimeout(10, TimeUnit.SECONDS).build().get();

consumer1.acknowledgeAsync(message.getMessageId(), transaction);

producer.newMessage(transaction).sendAsync(message.getValue());
producer1.newMessage(transaction).sendAsync(message.getValue());
producer2.newMessage(transaction).sendAsync(message.getValue());

transaction.commit().get();

} catch (Exception ignored) {

}
```
This is transaction commit internal code.
```
public CompletableFuture commit() {
return checkIfOpenOrCommitting().thenCompose((value) -> {
CompletableFuture 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;
});
}
```
It has the following problems :
**problem 1:** There are two exceptions in the code shown above when new transaction and end transaction. But we don't know what exception that operation will throw, if we new transaction throw exception, we should new transaction again. If we commit transaction throw exception, we should commit again.

**problem 2:** As we can see transaction commit internal code. When do transaction synchronization operation completed (ack with transaction and produce with transaction) then can commit this transaction. Although ACK and produce throw exceptions, we can re ACK and produce exceptions such as throw timeoutException. It will not affect the excetly-once semantics of the transaction. Because produce transaction message we can use Message deduplication and the exception thrown by ACK message is not Transactionconflictexception we can ack again.

**problem 3:** When transaction op fails, If you commit transaction, client will call abort. Such behavior will make users confused. Why do I call commit and client abort this transaction directly.
# Approach

## Exception catch

| Transaction op | Exception need to catch |
| ---- | ---- |
| New Transaction | NewTransactionTimeoutException |
| Produce Message With Transaction | InvalidTxnStatusException |
| Ack Message With Transaction | TransactionConflictException, InvalidTxnStatusException |
| End Transaction | TransactionNotFoundException, InvalidTxnStatusException, EndTransactionTimeoutException |

note:
1. Any other exception about transaction need to delete or change
2. Move all exception about transaction to extends PulsarClientException
## solve problem 1 :Handling exceptions of new transaction and commit transaction
new transaction : The only thing that can be determined here is that we are sure to throw an exception when the pulsar client is closed or the broker does not support transaction.
commit or abort transaction : When client closed and broker disable transaction or TransactionNotFoundException or InvalidTxnStatusException, user need to handle, and any another exception we can retry.
### All exceptions are thrown
Pulsar client don't need to handle any exceptions, throw all exceptions.
### Client handle the exception and retry create transaction
Because the client is not closed and the broker enable the transaction, we should not create transaction fail. The advantage of this is that no retryable exception will be thrown during the create transaction. The disadvantage is that users are not flexible enough to handle exceptions
### Summary
When client closed and broker disable transaction or TransactionNotFoundException or InvalidTxnStatusException, in this situation we can throw Exception to user, and user can handle this transaction. Users don't need to handle any other exceptions, because they don't care about other exceptions and have nothing to do with the user's use. Therefore, if we encounter other exceptions, we should automatically retry on the client side, which can greatly simplify the exception handling required by users using transaction
## solve problem 2 : Handle produce and ack exception
The exceptions that cannot be retried at all are client close or broker not opening transaction, and AckConflicException.
### Exception wrap
We wrap all retryable exceptions and let the client make the decision. like throw new TransactionOpFailException(e)
#### advantage
the user only needs to handle one exception. when do transaction op throw this exception, we can abort this transaction
#### disadvantage
users are not flexible enough to handle real exceptions, like produce or ack timeout we can try again In some specific context.
### Don't wrap
#### advantage
The user can do different op when throw different exception
#### disadvantage
The user may need to handle multiple exceptions and do different processing, which increases the complexity, but they can also abort by throwing exceptions.
### Summary
I think we should not wrap Exception, because in some special scenarios, a lot of messages may be sent, but a timeout in the middle will cause the transaction to be aborted, and the user does not have the change to choose to retry, which may lead to inflexibility and the user experience is not very good.
## solve problem 3 : Don't abort when user invoke commit
If we abort transaction when user invoke commit transaction, it will make users very confused.
### API Changes
User use API changed:
#### Before
```
consumer.receiveAsync().thenAccept(message -> {
try {
pulsarClient.newTransaction().withTransactionTimeout(10, TimeUnit.SECONDS).build().thenAccept(txn -> {
producer.newMessage(txn).sendAsync();
consumer.acknowledgeAsync(message.getMessageId());
txn.commit();
});
} catch (PulsarClientException pulsarClientException) {
pulsarClientException.printStackTrace();
}
});
```
users need to handle exception and don't know what time need to abort.
users can't handle exception when produce or ack message, because commit check all op(produce or ack) process success
users need catch pulsarClientException when use asynchronous API, because pulsarClient.newTransaction check the client if enable transaction
#### After
```
consumer.receiveAsync().thenAccept(message -> {
pulsarClient.newTransaction().withTransactionTimeout(10, TimeUnit.SECONDS).build().thenAccept(txn -> {
List> list = new ArrayList<>();
list.add(producer.newMessage(txn).sendAsync());
list.add(consumer.acknowledgeAsync(message.getMessageId()));
FutureUtil.waitForAll(list).thenAccept(v -> {
txn.commit().thenAccept(f -> {
log.info("Transaction commit success! txnId : {}", txn.getTxnID());
}).exceptionally(e -> {
if (e.getCause() instanceof TransactionNotFoundException) {
log.error("Transaction have been end! txnId : {}", txn.getTxnID(), e.getCause());
} else if (e.getCause() instanceof TransactionCoordinatorClientException.InvalidTxnStatusException) {
log.info("transaction is ending! txnId : {}, txnStatus : {}", txn.getTxnID(), ((TransactionImpl) txn).getState());
} else {
log.error("Client close or client disable transaction or broker disable transaction", e.getCause());
}
return null;
});
}).exceptionally(e -> {
txn.abort();
return null;
});
}).exceptionally(e -> {
log.error("Client close or client disable transaction or broker disable transaction", e.getCause());
return null;
});
});
```
As we can see, dealing with futures by users will make the process very complicated. If user don't handle transaction exception by themself when they do transaction op throw exception. We can provide a TransactionUtil to handle this situation.
```
public class TransactionUtil {

public static CompletableFuture prepareCommit(Transaction transaction) {
if (!(transaction instanceof TransactionImpl)) {
return FutureUtil.failedFuture(new IllegalArgumentException("Only support transactionImpl!"));
}
return ((TransactionImpl) transaction).prepareCommit();
}

}
```
If we use transactionUtil, the code is:
```
consumer.receiveAsync().thenAccept(message -> {
pulsarClient.newTransaction().withTransactionTimeout(10, TimeUnit.SECONDS).build().thenAccept(txn -> {
producer.newMessage(txn).sendAsync();
consumer.acknowledgeAsync(message.getMessageId());
TransactionUtil.prepareCommit(txn).thenAccept(v -> {
txn.commit().thenAccept(f -> {
log.info("Transaction commit success! txnId : {}", txn.getTxnID());
}).exceptionally(e -> {
if (e.getCause() instanceof TransactionNotFoundException) {
log.error("Transaction have been end! txnId : {}", txn.getTxnID(), e.getCause());
} else if (e.getCause() instanceof TransactionCoordinatorClientException.InvalidTxnStatusException) {
log.info("transaction is ending! txnId : {}, txnStatus : {}", txn.getTxnID(), ((TransactionImpl) txn).getState());
} else {
log.error("Client close or client disable transaction or broker disable transaction", e.getCause());
}
return null;
});
}).exceptionally(e -> {
txn.abort();
return null;
});
}).exceptionally(e -> {
log.error("Client close or client disable transaction or broker disable transaction", e.getCause());
return null;
});
});
```
If we use synchronous method, code like:
```
Message message = consumer.receive();
Transaction txn = null;
try {
txn = pulsarClient.newTransaction().withTransactionTimeout(10, TimeUnit.SECONDS).build().get();
producer.newMessage(txn).sendAsync();
consumer.acknowledgeAsync(message.getMessageId());
try {
TransactionUtil.prepareCommit(txn).get();
} catch (InterruptedException | ExecutionException e) {
txn.abort();
return;
}
txn.commit().get();
} catch (InterruptedException | ExecutionException e) {
if (txn != null) {
if (e.getCause() instanceof TransactionNotFoundException) {
log.error("Transaction have been end! txnId : {}", txn.getTxnID(), e.getCause());
} else if (e.getCause() instanceof TransactionCoordinatorClientException.InvalidTxnStatusException) {
log.info("transaction is ending! txnId : {}, txnStatus : {}", txn.getTxnID(), ((TransactionImpl) txn).getState());
} else {
log.error("Client close or client disable transaction or broker disable transaction", e.getCause());
}
} else {
log.error("Client close or client disable transaction or broker disable transaction", e.getCause());
}
}
```

# Documentation Changes [Optional]
https://pulsar.incubator.apache.org/docs/en/txn-use/ should change use example code.
# Implementation

## TransactionCoordinatorClientException
| TransactionCoordinatorClientException | operation |
| ---- | ---- |
| CoordinatorClientStateException | delete |
| CoordinatorNotFoundException | move to PulsarClientException |
| TransactionNotFoundException | move to PulsarClientException |
| InvalidTxnStatusException | move to PulsarClientException |
| MetaStoreHandlerNotExistsException | move to PulsarClientException |
| MetaStoreHandlerNotReadyException | move to PulsarClientException, when TransactionMetaStoreHandler state is close or closing we should throw MetaStoreHandlerHasClosedException, create a new exception `MetaStoreHandlerHasClosedException` |

## TransactionCoordinatorClientImpl
```
@Override
public void commit(TxnID txnID) throws TransactionCoordinatorClientException {
try {
commitAsync(txnID).get();
} catch (Exception e) {
throw TransactionCoordinatorClientException.unwrap(e);
}
}
```
change to

```
@Override
public void commit(TxnID txnID) throws TransactionCoordinatorClientException {
try {
commitAsync(txnID).get();
} catch (Exception e) {
throw TransactionCoordinatorClientException.unwrap(e.getCause());
}
}
```
same like `addSubscriptionToTxn` `addPublishPartitionToTxn` `newTransaction`

## If client or broker disable transaction throw exception
All throw PulsarClientException.NotAllowedException

## New transaction retry
we should retry the op when receive exception except PulsarClientException.NotAllowedException or MetaStoreHandlerHasClosedException and NewTransactionTimeoutException. Add Exception `NewTransactionTimeoutException` when new transaction timeout.

## End transaction retry
We should retry the op when receive exception except PulsarClientException.NotAllowedException or MetaStoreHandlerHasClosedException and TransactionNotFoundException and InvalidTxnStatusException and EndTransactionTimeoutException. Add Exception `EndTransactionTimeoutException` when new transaction timeout.

## Commit or abort don't wait all transaction op completely
now commit wait every transaction op complete, we don't need to use this logical.

## Commit will no invoke abort
delete the logic when commit invoke abort.

## Add TransactionUtil to Simplify user steps

```
public class TransactionUtil {

public static CompletableFuture prepareCommit(Transaction transaction) {
if (!(transaction instanceof TransactionImpl)) {
return FutureUtil.failedFuture(new IllegalArgumentException("Only support transactionImpl!"));
}
return ((TransactionImpl) transaction).prepareCommit();
}
}
```
# Compatibility
user invoke commit will not abort any transaction.
user need to handle transaction op(produce or ack) by themself.
# Test Plan [Optional]
new and end transaction retry test
test commit will not check transaction op

Start with the transaction API entry points shown in the issue: newTransaction(), TransactionImpl.commit(), abort(), and the proposed TransactionUtil.prepareCommit(). Review the listed exception cases and asynchronous examples, then compare them with the current transaction implementation. Done means the commit, abort, retry, and exception-handling semantics are agreed and reflected in the client API.

Written by the indexing model from the issue text.