syhily commented on code in PR #17718:
URL: https://github.com/apache/pulsar/pull/17718#discussion_r974760724
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/TransactionImpl.java:
##########
@@ -133,15 +133,9 @@ public CompletableFuture<Void> registerAckedTopic(String
topic, String subscript
if (checkIfOpen(completableFuture)) {
synchronized (TransactionImpl.this) {
// we need to issue the request to TC to register the acked
topic
- return registerSubscriptionMap.compute(Pair.of(topic,
subscription), (key, future) -> {
- if (future != null) {
- return future.thenCompose(ignored ->
CompletableFuture.completedFuture(null));
- } else {
- return tcClient.addSubscriptionToTxnAsync(
- new TxnID(txnIdMostBits, txnIdLeastBits),
topic, subscription)
- .thenCompose(ignored ->
CompletableFuture.completedFuture(null));
- }
- });
+ return registerSubscriptionMap.computeIfAbsent(Pair.of(topic,
subscription), key ->
+ tcClient.addSubscriptionToTxnAsync(
+ new TxnID(txnIdMostBits, txnIdLeastBits),
topic, subscription));
}
}
return completableFuture;
Review Comment:
No need to `checkIfOpen`, because it will always be open.
```
public CompletableFuture<Void> registerAckedTopic(String topic, String
subscription) {
CompletableFuture<Void> completableFuture = new
CompletableFuture<>();
synchronized (TransactionImpl.this) {
// we need to issue the request to TC to register the acked topic
return registerSubscriptionMap.computeIfAbsent(Pair.of(topic,
subscription), key ->
tcClient.addSubscriptionToTxnAsync(new
TxnID(txnIdMostBits, txnIdLeastBits), topic, subscription));
}
}
```
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/TransactionImpl.java:
##########
@@ -133,15 +133,9 @@ public CompletableFuture<Void> registerAckedTopic(String
topic, String subscript
if (checkIfOpen(completableFuture)) {
synchronized (TransactionImpl.this) {
// we need to issue the request to TC to register the acked
topic
- return registerSubscriptionMap.compute(Pair.of(topic,
subscription), (key, future) -> {
- if (future != null) {
- return future.thenCompose(ignored ->
CompletableFuture.completedFuture(null));
- } else {
- return tcClient.addSubscriptionToTxnAsync(
- new TxnID(txnIdMostBits, txnIdLeastBits),
topic, subscription)
- .thenCompose(ignored ->
CompletableFuture.completedFuture(null));
- }
- });
+ return registerSubscriptionMap.computeIfAbsent(Pair.of(topic,
subscription), key ->
+ tcClient.addSubscriptionToTxnAsync(
+ new TxnID(txnIdMostBits, txnIdLeastBits),
topic, subscription));
}
}
return completableFuture;
Review Comment:
No need to `checkIfOpen`, because it will always be open.
```
public CompletableFuture<Void> registerAckedTopic(String topic, String
subscription) {
CompletableFuture<Void> completableFuture = new CompletableFuture<>();
synchronized (TransactionImpl.this) {
// we need to issue the request to TC to register the acked topic
return registerSubscriptionMap.computeIfAbsent(Pair.of(topic,
subscription), key ->
tcClient.addSubscriptionToTxnAsync(new TxnID(txnIdMostBits,
txnIdLeastBits), topic, subscription));
}
}
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
