[Transaction] Fix transaction buffer client handle endTxn op when topic or sub have been deleted. #11304
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -23,6 +23,7 @@ | |
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.Optional; | ||
| import java.util.concurrent.CompletableFuture; | ||
| import java.util.concurrent.CompletionException; | ||
|
|
||
| import lombok.extern.slf4j.Slf4j; | ||
|
|
@@ -229,6 +230,11 @@ public Iterable<String> getManagedLedgers() throws MetaStoreException { | |
| } | ||
| } | ||
|
|
||
| @Override | ||
| public CompletableFuture<Boolean> exists(String path) { | ||
| return store.exists(PREFIX + path); | ||
| } | ||
|
|
||
| // | ||
| // update timestamp if missing or 0 | ||
| // 3 cases - timestamp does not exist for ledgers serialized before | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -1925,43 +1925,68 @@ protected void handleEndTxnOnPartition(CommandEndTxnOnPartition command) { | |
| final String topic = command.getTopic(); | ||
| final int txnAction = command.getTxnAction().getValue(); | ||
| TxnID txnID = new TxnID(command.getTxnidMostBits(), command.getTxnidLeastBits()); | ||
| final long lowWaterMark = command.getTxnidLeastBitsOfLowWatermark(); | ||
|
|
||
| if (log.isDebugEnabled()) { | ||
| log.debug("[{}] handleEndTxnOnPartition txnId: [{}], txnAction: [{}]", topic, | ||
| txnID, txnAction); | ||
| } | ||
| CompletableFuture<Optional<Topic>> topicFuture = service.getTopicIfExists(TopicName.get(topic).toString()); | ||
| topicFuture.thenAccept(optionalTopic -> { | ||
| if (optionalTopic.isPresent()) { | ||
| optionalTopic.get().endTxn(txnID, txnAction, lowWaterMark) | ||
| .whenComplete((ignored, throwable) -> { | ||
| if (throwable != null) { | ||
| log.error("handleEndTxnOnPartition fail!, topic {}, txnId: [{}], " | ||
| + "txnAction: [{}]", topic, txnID, TxnAction.valueOf(txnAction), throwable); | ||
| ctx.writeAndFlush(Commands.newEndTxnOnPartitionResponse( | ||
| requestId, BrokerServiceException.getClientErrorCode(throwable), | ||
| throwable.getMessage(), | ||
| txnID.getLeastSigBits(), txnID.getMostSigBits())); | ||
| return; | ||
| } | ||
| ctx.writeAndFlush(Commands.newEndTxnOnPartitionResponse(requestId, | ||
| txnID.getLeastSigBits(), txnID.getMostSigBits())); | ||
| }); | ||
|
|
||
| } else { | ||
| getBrokerService().getManagedLedgerFactory() | ||
| .checkManagedLedgerInitializedBefore(TopicName.get(topic).getPersistenceNamingEncoding()) | ||
| .thenAccept((b) -> { | ||
| if (b) { | ||
| log.error("handleEndTxnOnPartition fail ! The topic {} does not exist in broker, " | ||
| + "txnId: [{}], txnAction: [{}]", topic, | ||
| txnID, TxnAction.valueOf(txnAction)); | ||
| ctx.writeAndFlush(Commands.newEndTxnOnPartitionResponse(requestId, | ||
| ServerError.ServiceNotReady, | ||
| "The topic " + topic + " is not exist in broker.", | ||
|
There was a problem hiding this comment. there is a typo, what about: "The topic " + topic + " does not exist in broker." |
||
| txnID.getMostSigBits(), txnID.getLeastSigBits())); | ||
| } else { | ||
| log.warn("handleEndTxnOnPartition fail ! The topic {} have not been created, " | ||
|
There was a problem hiding this comment. typo: "has not been created" |
||
| + "txnId: [{}], txnAction: [{}]", | ||
| topic, txnID, TxnAction.valueOf(txnAction)); | ||
| ctx.writeAndFlush(Commands.newEndTxnOnPartitionResponse(requestId, | ||
| txnID.getLeastSigBits(), txnID.getMostSigBits())); | ||
| } | ||
| }).exceptionally(e -> { | ||
| log.error("handleEndTxnOnPartition fail ! topic {} , " | ||
| + "txnId: [{}], txnAction: [{}]", topic, txnID, | ||
| TxnAction.valueOf(txnAction), e.getCause()); | ||
| ctx.writeAndFlush(Commands.newEndTxnOnPartitionResponse( | ||
| requestId, ServerError.ServiceNotReady, | ||
| e.getMessage(), txnID.getLeastSigBits(), txnID.getMostSigBits())); | ||
| return null; | ||
| }); | ||
| } | ||
| }).exceptionally(e -> { | ||
| log.error("handleEndTxnOnPartition fail ! topic {} , " | ||
| + "txnId: [{}], txnAction: [{}]", topic, txnID, | ||
| TxnAction.valueOf(txnAction), e.getCause()); | ||
| ctx.writeAndFlush(Commands.newEndTxnOnPartitionResponse( | ||
| requestId, ServerError.ServiceNotReady, | ||
| e.getMessage(), txnID.getLeastSigBits(), txnID.getMostSigBits())); | ||
| return null; | ||
| }); | ||
| } | ||
|
|
||
| @Override | ||
|
|
@@ -1972,70 +1997,82 @@ protected void handleEndTxnOnSubscription(CommandEndTxnOnSubscription command) { | |
| final String topic = command.getSubscription().getTopic(); | ||
| final String subName = command.getSubscription().getSubscription(); | ||
| final int txnAction = command.getTxnAction().getValue(); | ||
| final TxnID txnID = new TxnID(txnidMostBits, txnidLeastBits); | ||
| final long lowWaterMark = command.getTxnidLeastBitsOfLowWatermark(); | ||
|
|
||
| if (log.isDebugEnabled()) { | ||
| log.debug("[{}] [{}] handleEndTxnOnSubscription txnId: [{}], txnAction: [{}]", topic, subName, | ||
| new TxnID(txnidMostBits, txnidLeastBits), txnAction); | ||
| } | ||
|
|
||
| CompletableFuture<Optional<Topic>> topicFuture = service.getTopicIfExists(TopicName.get(topic).toString()); | ||
| topicFuture.thenAccept(optionalTopic -> { | ||
| if (optionalTopic.isPresent()) { | ||
| Subscription subscription = optionalTopic.get().getSubscription(subName); | ||
| if (subscription == null) { | ||
| log.warn("handleEndTxnOnSubscription fail! " | ||
| + "topic {} subscription {} is not exist. txnId: [{}], txnAction: [{}]", | ||
|
There was a problem hiding this comment. "is not exist" -> "does not exist" |
||
| optionalTopic.get().getName(), subName, txnID, TxnAction.valueOf(txnAction)); | ||
| ctx.writeAndFlush( | ||
| Commands.newEndTxnOnSubscriptionResponse(requestId, txnidLeastBits, txnidMostBits)); | ||
| return; | ||
| } | ||
|
|
||
| CompletableFuture<Void> completableFuture = | ||
| subscription.endTxn(txnidMostBits, txnidLeastBits, txnAction, lowWaterMark); | ||
| completableFuture.whenComplete((ignored, e) -> { | ||
| if (e != null) { | ||
| log.error("handleEndTxnOnSubscription fail ! topic: {} , subscription: {}" | ||
| + "txnId: [{}], txnAction: [{}]", topic, subName, | ||
| txnID, TxnAction.valueOf(txnAction), e.getCause()); | ||
| ctx.writeAndFlush(Commands.newEndTxnOnSubscriptionResponse( | ||
| requestId, txnidLeastBits, txnidMostBits, | ||
| BrokerServiceException.getClientErrorCode(e), | ||
| "Handle end txn on subscription failed.")); | ||
| return; | ||
| } | ||
| ctx.writeAndFlush( | ||
| Commands.newEndTxnOnSubscriptionResponse(requestId, txnidLeastBits, txnidMostBits)); | ||
| }); | ||
| } else { | ||
| getBrokerService().getManagedLedgerFactory() | ||
| .checkManagedLedgerInitializedBefore(TopicName.get(topic).getPersistenceNamingEncoding()) | ||
| .thenAccept((b) -> { | ||
| if (b) { | ||
| log.error("handleEndTxnOnSubscription fail! The topic {} does not exist in broker, " | ||
| + "subscription: {} ,txnId: [{}], txnAction: [{}]", topic, subName, | ||
| new TxnID(txnidMostBits, txnidLeastBits), TxnAction.valueOf(txnAction)); | ||
| ctx.writeAndFlush(Commands.newEndTxnOnSubscriptionResponse( | ||
| requestId, txnID.getLeastSigBits(), txnID.getMostSigBits(), | ||
| ServerError.ServiceNotReady, | ||
| "The topic " + topic + " is not exist in broker.")); | ||
| } else { | ||
| log.warn("handleEndTxnOnSubscription fail ! The topic {} have not been created, " | ||
| + "subscription: {} txnId: [{}], txnAction: [{}]", | ||
| topic, subName, txnID, TxnAction.valueOf(txnAction)); | ||
| ctx.writeAndFlush(Commands.newEndTxnOnSubscriptionResponse(requestId, | ||
| txnID.getLeastSigBits(), txnID.getMostSigBits())); | ||
| } | ||
| }).exceptionally(e -> { | ||
| log.error("handleEndTxnOnSubscription fail ! topic {} , subscription: {}" | ||
| + "txnId: [{}], txnAction: [{}]", topic, subName, | ||
| txnID, TxnAction.valueOf(txnAction), e.getCause()); | ||
| ctx.writeAndFlush(Commands.newEndTxnOnSubscriptionResponse( | ||
| requestId, txnID.getLeastSigBits(), txnID.getMostSigBits(), | ||
| ServerError.ServiceNotReady, e.getMessage())); | ||
| return null; | ||
| }); | ||
| } | ||
| }).exceptionally(e -> { | ||
| log.error("handleEndTxnOnSubscription fail ! topic: {} , subscription: {}" | ||
| + "txnId: [{}], txnAction: [{}]", topic, subName, | ||
| txnID, TxnAction.valueOf(txnAction), e.getCause()); | ||
| ctx.writeAndFlush(Commands.newEndTxnOnSubscriptionResponse( | ||
| requestId, txnidLeastBits, txnidMostBits, | ||
| ServerError.ServiceNotReady, | ||
| "Handle end txn on subscription failed.")); | ||
| return null; | ||
| }); | ||
| } | ||
|
|
||
| private CompletableFuture<SchemaVersion> tryAddSchema(Topic topic, SchemaData schema) { | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
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.