[Broker] Fix call sync method in async rest api for internalGetMessageById.
@@ -2390,19 +2390,27 @@ public class PersistentTopicsBase extends AdminResource {
protected void internalGetMessageById(AsyncResponse asyncResponse, long ledgerId, long entryId,
boolean authoritative) {
- try {
// will redirect if the topic not owned by current broker
- validateTopicOwnership(topicName, authoritative);
- validateTopicOperation(topicName, TopicOperation.PEEK_MESSAGES);
-
+ validateTopicOwnershipAsync(topicName, authoritative)
+ .thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES))
+ .thenCompose(__ -> {
+ CompletableFuture<Void> ret;
if (topicName.isGlobal()) {
- validateGlobalNamespaceOwnership(namespaceName);
+ ret = validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ ret = CompletableFuture.completedFuture(null);
}
- PersistentTopic topic = (PersistentTopic) getTopicReference(topicName);
- ManagedLedgerImpl ledger = (ManagedLedgerImpl) topic.getManagedLedger();
- ledger.asyncReadEntry(new PositionImpl(ledgerId, entryId), new AsyncCallbacks.ReadEntryCallback() {
+ return ret;
+ })
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenAccept(topic -> {
+ ManagedLedgerImpl ledger =
+ (ManagedLedgerImpl) ((PersistentTopic) topic).getManagedLedger();
+ ledger.asyncReadEntry(new PositionImpl(ledgerId, entryId),
+ new AsyncCallbacks.ReadEntryCallback() {
@Override
- public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ public void readEntryFailed(ManagedLedgerException exception,
+ Object ctx) {
asyncResponse.resume(new RestException(exception));
}
@@ -2419,13 +2427,13 @@ public class PersistentTopicsBase extends AdminResource {
}
}
}, null);
- } catch (NullPointerException npe) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND, "Message not found"));
- } catch (Exception exception) {
+ }).exceptionally(ex -> {
+ Throwable cause = ex.getCause();
log.error("[{}] Failed to get message with ledgerId {} entryId {} from {}",
- clientAppId(), ledgerId, entryId, topicName, exception);
- asyncResponse.resume(new RestException(exception));
- }
+ clientAppId(), ledgerId, entryId, topicName, cause);
+ asyncResponse.resume(new RestException(cause));
+ return null;
+ });
}
protected CompletableFuture<MessageId> internalGetMessageIdByTimestamp(long timestamp, boolean authoritative) {
