[Broker] Fix call sync method in async rest api for internalGetLastMessageId
@@ -4169,31 +4169,20 @@ public class PersistentTopicsBase extends AdminResource {
}
protected void internalGetLastMessageId(AsyncResponse asyncResponse, boolean authoritative) {
- Topic topic;
- try {
- validateTopicOwnership(topicName, authoritative);
- validateTopicOperation(topicName, TopicOperation.PEEK_MESSAGES);
- topic = getTopicReference(topicName);
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to get last messageId {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- return;
- } catch (Exception e) {
- log.error("[{}] Failed to get last messageId {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
+ validateTopicOwnershipAsync(topicName, authoritative)
+ .thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenAccept(topic -> {
+ if (topic == null) {
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, "Topic not found"));
return;
}
-
if (!(topic instanceof PersistentTopic)) {
log.error("[{}] Not supported operation of non-persistent topic {}", clientAppId(), topicName);
asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED,
"GetLastMessageId on a non-persistent topic is not allowed"));
return;
}
-
topic.getLastMessageId().whenComplete((v, e) -> {
if (e != null) {
asyncResponse.resume(new RestException(Status.INTERNAL_SERVER_ERROR, e.getMessage()));
@@ -4201,7 +4190,12 @@ public class PersistentTopicsBase extends AdminResource {
asyncResponse.resume(v);
}
});
-
+ }).exceptionally(ex -> {
+ Throwable cause = FutureUtil.unwrapCompletionException(ex);
+ log.error("[{}] Failed to get last messageId {}", clientAppId(), topicName, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, cause);
+ return null;
+ });
}
protected CompletableFuture<DispatchRateImpl> internalGetDispatchRate(boolean applied, boolean isGlobal) {
