You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/01/21 06:12:20 UTC
[GitHub] [pulsar] mattisonchao commented on a change in pull request #13871: [Broker] Fix call sync method in async rest api for internalGetBacklogSizeByMessageId
mattisonchao commented on a change in pull request #13871:
URL: https://github.com/apache/pulsar/pull/13871#discussion_r789383715
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -2783,49 +2783,55 @@ protected PersistentOfflineTopicStats internalGetBacklog(boolean authoritative)
protected void internalGetBacklogSizeByMessageId(AsyncResponse asyncResponse,
MessageIdImpl messageId, boolean authoritative) {
+ CompletableFuture<Void> future;
if (topicName.isGlobal()) {
- try {
- validateGlobalNamespaceOwnership(namespaceName);
- } catch (Exception e) {
- log.error("[{}] Failed to get backlog size for topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- }
- PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(topicName,
- authoritative, false);
- if (!topicName.isPartitioned() && partitionMetadata.partitions > 0) {
- log.warn("[{}] Not supported calculate backlog size operation on partitioned-topic {}",
- clientAppId(), topicName);
- asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED,
- "calculate backlog size is not allowed for partitioned-topic"));
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
} else {
- validateTopicOwnership(topicName, authoritative);
- validateTopicOperation(topicName, TopicOperation.GET_BACKLOG_SIZE);
- PersistentTopic topic = (PersistentTopic) getTopicReference(topicName);
- PositionImpl pos = new PositionImpl(messageId.getLedgerId(), messageId.getEntryId());
- if (topic == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND, "Topic not found"));
- return;
- }
- try {
- ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) topic.getManagedLedger();
- if (messageId.getLedgerId() == -1) {
- asyncResponse.resume(managedLedger.getTotalSize());
- } else {
- asyncResponse.resume(managedLedger.getEstimatedBacklogSize(pos));
- }
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to get backlog size for topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- } catch (Exception e) {
- log.error("[{}] Failed to get backlog size for topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- }
+ future = CompletableFuture.completedFuture(null);
}
+ future.thenAccept(__ -> {
+ getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(partitionMetadata -> {
+ if (!topicName.isPartitioned() && partitionMetadata.partitions > 0) {
+ log.warn("[{}] Not supported calculate backlog size operation on partitioned-topic {}",
+ clientAppId(), topicName);
+ asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED,
+ "calculate backlog size is not allowed for partitioned-topic"));
+ } else {
+ validateTopicOwnership(topicName, authoritative);
Review comment:
this is a sync method need to fix.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -2783,49 +2783,55 @@ protected PersistentOfflineTopicStats internalGetBacklog(boolean authoritative)
protected void internalGetBacklogSizeByMessageId(AsyncResponse asyncResponse,
MessageIdImpl messageId, boolean authoritative) {
+ CompletableFuture<Void> future;
if (topicName.isGlobal()) {
- try {
- validateGlobalNamespaceOwnership(namespaceName);
- } catch (Exception e) {
- log.error("[{}] Failed to get backlog size for topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- }
- PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(topicName,
- authoritative, false);
- if (!topicName.isPartitioned() && partitionMetadata.partitions > 0) {
- log.warn("[{}] Not supported calculate backlog size operation on partitioned-topic {}",
- clientAppId(), topicName);
- asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED,
- "calculate backlog size is not allowed for partitioned-topic"));
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
Review comment:
We need to handle validate global namespace ownership exception.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -2783,49 +2783,55 @@ protected PersistentOfflineTopicStats internalGetBacklog(boolean authoritative)
protected void internalGetBacklogSizeByMessageId(AsyncResponse asyncResponse,
MessageIdImpl messageId, boolean authoritative) {
+ CompletableFuture<Void> future;
if (topicName.isGlobal()) {
- try {
- validateGlobalNamespaceOwnership(namespaceName);
- } catch (Exception e) {
- log.error("[{}] Failed to get backlog size for topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- }
- PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(topicName,
- authoritative, false);
- if (!topicName.isPartitioned() && partitionMetadata.partitions > 0) {
- log.warn("[{}] Not supported calculate backlog size operation on partitioned-topic {}",
- clientAppId(), topicName);
- asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED,
- "calculate backlog size is not allowed for partitioned-topic"));
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
} else {
- validateTopicOwnership(topicName, authoritative);
- validateTopicOperation(topicName, TopicOperation.GET_BACKLOG_SIZE);
- PersistentTopic topic = (PersistentTopic) getTopicReference(topicName);
- PositionImpl pos = new PositionImpl(messageId.getLedgerId(), messageId.getEntryId());
- if (topic == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND, "Topic not found"));
- return;
- }
- try {
- ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) topic.getManagedLedger();
- if (messageId.getLedgerId() == -1) {
- asyncResponse.resume(managedLedger.getTotalSize());
- } else {
- asyncResponse.resume(managedLedger.getEstimatedBacklogSize(pos));
- }
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to get backlog size for topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- } catch (Exception e) {
- log.error("[{}] Failed to get backlog size for topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- }
+ future = CompletableFuture.completedFuture(null);
}
+ future.thenAccept(__ -> {
+ getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(partitionMetadata -> {
+ if (!topicName.isPartitioned() && partitionMetadata.partitions > 0) {
+ log.warn("[{}] Not supported calculate backlog size operation on partitioned-topic {}",
+ clientAppId(), topicName);
+ asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED,
+ "calculate backlog size is not allowed for partitioned-topic"));
+ } else {
+ validateTopicOwnership(topicName, authoritative);
+ validateTopicOperation(topicName, TopicOperation.GET_BACKLOG_SIZE);
+ PersistentTopic topic = (PersistentTopic) getTopicReference(topicName);
Review comment:
same above.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -2783,49 +2783,55 @@ protected PersistentOfflineTopicStats internalGetBacklog(boolean authoritative)
protected void internalGetBacklogSizeByMessageId(AsyncResponse asyncResponse,
MessageIdImpl messageId, boolean authoritative) {
+ CompletableFuture<Void> future;
if (topicName.isGlobal()) {
- try {
- validateGlobalNamespaceOwnership(namespaceName);
- } catch (Exception e) {
- log.error("[{}] Failed to get backlog size for topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- }
- PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(topicName,
- authoritative, false);
- if (!topicName.isPartitioned() && partitionMetadata.partitions > 0) {
- log.warn("[{}] Not supported calculate backlog size operation on partitioned-topic {}",
- clientAppId(), topicName);
- asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED,
- "calculate backlog size is not allowed for partitioned-topic"));
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
} else {
- validateTopicOwnership(topicName, authoritative);
- validateTopicOperation(topicName, TopicOperation.GET_BACKLOG_SIZE);
- PersistentTopic topic = (PersistentTopic) getTopicReference(topicName);
- PositionImpl pos = new PositionImpl(messageId.getLedgerId(), messageId.getEntryId());
- if (topic == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND, "Topic not found"));
- return;
- }
- try {
- ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) topic.getManagedLedger();
- if (messageId.getLedgerId() == -1) {
- asyncResponse.resume(managedLedger.getTotalSize());
- } else {
- asyncResponse.resume(managedLedger.getEstimatedBacklogSize(pos));
- }
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to get backlog size for topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- } catch (Exception e) {
- log.error("[{}] Failed to get backlog size for topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- }
+ future = CompletableFuture.completedFuture(null);
}
+ future.thenAccept(__ -> {
+ getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(partitionMetadata -> {
+ if (!topicName.isPartitioned() && partitionMetadata.partitions > 0) {
+ log.warn("[{}] Not supported calculate backlog size operation on partitioned-topic {}",
+ clientAppId(), topicName);
+ asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED,
+ "calculate backlog size is not allowed for partitioned-topic"));
+ } else {
+ validateTopicOwnership(topicName, authoritative);
+ validateTopicOperation(topicName, TopicOperation.GET_BACKLOG_SIZE);
+ PersistentTopic topic = (PersistentTopic) getTopicReference(topicName);
+ PositionImpl pos = new PositionImpl(messageId.getLedgerId(), messageId.getEntryId());
+ if (topic == null) {
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, "Topic not found"));
+ return;
+ }
+ try {
+ ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) topic.getManagedLedger();
+ if (messageId.getLedgerId() == -1) {
+ asyncResponse.resume(managedLedger.getTotalSize());
+ } else {
+ asyncResponse.resume(managedLedger.getEstimatedBacklogSize(pos));
+ }
+ } catch (WebApplicationException wae) {
Review comment:
Why we catch here ?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -2783,49 +2783,55 @@ protected PersistentOfflineTopicStats internalGetBacklog(boolean authoritative)
protected void internalGetBacklogSizeByMessageId(AsyncResponse asyncResponse,
MessageIdImpl messageId, boolean authoritative) {
+ CompletableFuture<Void> future;
if (topicName.isGlobal()) {
- try {
- validateGlobalNamespaceOwnership(namespaceName);
- } catch (Exception e) {
- log.error("[{}] Failed to get backlog size for topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- }
- PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(topicName,
- authoritative, false);
- if (!topicName.isPartitioned() && partitionMetadata.partitions > 0) {
- log.warn("[{}] Not supported calculate backlog size operation on partitioned-topic {}",
- clientAppId(), topicName);
- asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED,
- "calculate backlog size is not allowed for partitioned-topic"));
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
} else {
- validateTopicOwnership(topicName, authoritative);
- validateTopicOperation(topicName, TopicOperation.GET_BACKLOG_SIZE);
- PersistentTopic topic = (PersistentTopic) getTopicReference(topicName);
- PositionImpl pos = new PositionImpl(messageId.getLedgerId(), messageId.getEntryId());
- if (topic == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND, "Topic not found"));
- return;
- }
- try {
- ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) topic.getManagedLedger();
- if (messageId.getLedgerId() == -1) {
- asyncResponse.resume(managedLedger.getTotalSize());
- } else {
- asyncResponse.resume(managedLedger.getEstimatedBacklogSize(pos));
- }
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to get backlog size for topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- } catch (Exception e) {
- log.error("[{}] Failed to get backlog size for topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- }
+ future = CompletableFuture.completedFuture(null);
}
+ future.thenAccept(__ -> {
+ getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(partitionMetadata -> {
+ if (!topicName.isPartitioned() && partitionMetadata.partitions > 0) {
+ log.warn("[{}] Not supported calculate backlog size operation on partitioned-topic {}",
+ clientAppId(), topicName);
+ asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED,
+ "calculate backlog size is not allowed for partitioned-topic"));
+ } else {
+ validateTopicOwnership(topicName, authoritative);
+ validateTopicOperation(topicName, TopicOperation.GET_BACKLOG_SIZE);
Review comment:
same above.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org