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/23 12:00:27 UTC
[GitHub] [pulsar] Jason918 commented on a change in pull request #13905: [Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…
Jason918 commented on a change in pull request #13905:
URL: https://github.com/apache/pulsar/pull/13905#discussion_r790264679
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -1801,124 +1801,128 @@ protected void internalSkipMessages(String subName, int numMessages, boolean aut
}
protected void internalExpireMessagesForAllSubscriptions(AsyncResponse asyncResponse, int expireTimeInSeconds,
- boolean authoritative) {
+ boolean authoritative) {
+ CompletableFuture<Void> future;
if (topicName.isGlobal()) {
- try {
- validateGlobalNamespaceOwnership(namespaceName);
- } catch (Exception e) {
- log.error("[{}] Failed to expire messages for all subscription on topic {}",
- clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ future = CompletableFuture.completedFuture(null);
}
- // If the topic name is a partition name, no need to get partition topic metadata again
- if (topicName.isPartitioned()) {
- internalExpireMessagesForAllSubscriptionsForNonPartitionedTopic(asyncResponse,
+ future.thenCompose(__ -> {
+ // If the topic name is a partition name, no need to get partition topic metadata again
+ if (topicName.isPartitioned()) {
+ return internalExpireMessagesForAllSubscriptionsForNonPartitionedTopicAsync(asyncResponse,
expireTimeInSeconds, authoritative);
- } else {
- getPartitionedTopicMetadataAsync(topicName,
- authoritative, false).thenAccept(partitionMetadata -> {
- if (partitionMetadata.partitions > 0) {
- final List<CompletableFuture<Void>> futures = Lists.newArrayList();
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenCompose(partitionMetadata -> {
+ if (partitionMetadata.partitions > 0) {
+ final List<CompletableFuture<Void>> futures = Lists.newArrayList();
- // expire messages for each partition topic
- for (int i = 0; i < partitionMetadata.partitions; i++) {
- TopicName topicNamePartition = topicName.getPartition(i);
- try {
- futures.add(pulsar()
- .getAdminClient()
- .topics()
- .expireMessagesForAllSubscriptionsAsync(
+ // expire messages for each partition topic
+ for (int i = 0; i < partitionMetadata.partitions; i++) {
+ TopicName topicNamePartition = topicName.getPartition(i);
+ try {
+ futures.add(pulsar()
+ .getAdminClient()
+ .topics()
+ .expireMessagesForAllSubscriptionsAsync(
topicNamePartition.toString(), expireTimeInSeconds));
- } catch (Exception e) {
- log.error("[{}] Failed to expire messages up to {} on {}",
- clientAppId(), expireTimeInSeconds,
- topicNamePartition, e);
- asyncResponse.resume(new RestException(e));
- return;
- }
- }
+ } catch (Exception e) {
+ log.error("[{}] Failed to expire messages up to {} on {}",
+ clientAppId(), expireTimeInSeconds,
+ topicNamePartition, e);
+ asyncResponse.resume(new RestException(e));
+ return CompletableFuture.completedFuture(null);
+ }
+ }
- FutureUtil.waitForAll(futures).handle((result, exception) -> {
- if (exception != null) {
- Throwable t = exception.getCause();
- log.error("[{}] Failed to expire messages up to {} on {}",
- clientAppId(), expireTimeInSeconds,
- topicName, t);
- asyncResponse.resume(new RestException(t));
- return null;
+ return FutureUtil.waitForAll(futures).handle((result, exception) -> {
+ if (exception != null) {
+ Throwable t = exception.getCause();
+ log.error("[{}] Failed to expire messages up to {} on {}",
+ clientAppId(), expireTimeInSeconds,
+ topicName, t);
+ asyncResponse.resume(new RestException(t));
+ return null;
+ }
+ asyncResponse.resume(Response.noContent().build());
+ return null;
+ });
+ } else {
+ return internalExpireMessagesForAllSubscriptionsForNonPartitionedTopicAsync(asyncResponse,
+ expireTimeInSeconds, authoritative);
}
-
- asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(ex -> {
Review comment:
Should this `exceptionally` in Line 1863?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -3445,6 +3449,92 @@ private void internalExpireMessagesByTimestampForSinglePartition(String subName,
}
}
+ private CompletableFuture<Void> internalExpireMessagesByTimestampForSinglePartitionAsync(String subName,
+ int expireTimeInSeconds,
+ boolean authoritative) {
+ CompletableFuture<Void> future;
+ if (topicName.isGlobal()) {
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ future = CompletableFuture.completedFuture(null);
+ }
+
+ return future.thenCompose(__ -> {
+ // If the topic name is a partition name, no need to get partition topic metadata again
+ if (!topicName.isPartitioned()) {
+ getPartitionedTopicMetadataAsync(topicName, authoritative, false)
Review comment:
return?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -3445,6 +3449,92 @@ private void internalExpireMessagesByTimestampForSinglePartition(String subName,
}
}
+ private CompletableFuture<Void> internalExpireMessagesByTimestampForSinglePartitionAsync(String subName,
+ int expireTimeInSeconds,
+ boolean authoritative) {
+ CompletableFuture<Void> future;
+ if (topicName.isGlobal()) {
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ future = CompletableFuture.completedFuture(null);
+ }
+
+ return future.thenCompose(__ -> {
+ // If the topic name is a partition name, no need to get partition topic metadata again
+ if (!topicName.isPartitioned()) {
+ getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenCompose(metadata -> {
+ if (metadata.partitions > 0) {
+ String msg = "This method should not be called for partitioned topic";
+ log.error("[{}] {} {} {}", clientAppId(), msg, topicName, subName);
+ throw new IllegalStateException(msg);
+ } else {
+ return doInternalExpireMessagesByTimestampForSinglePartitionAsync(subName,
Review comment:
We can return a `CompletableFuture.completedFuture(null)` if this check is passed to avoid add another method.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -3445,6 +3449,92 @@ private void internalExpireMessagesByTimestampForSinglePartition(String subName,
}
}
+ private CompletableFuture<Void> internalExpireMessagesByTimestampForSinglePartitionAsync(String subName,
Review comment:
@liudezhi2098 Any difference between these two?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -1801,124 +1801,128 @@ protected void internalSkipMessages(String subName, int numMessages, boolean aut
}
protected void internalExpireMessagesForAllSubscriptions(AsyncResponse asyncResponse, int expireTimeInSeconds,
- boolean authoritative) {
+ boolean authoritative) {
+ CompletableFuture<Void> future;
if (topicName.isGlobal()) {
- try {
- validateGlobalNamespaceOwnership(namespaceName);
- } catch (Exception e) {
- log.error("[{}] Failed to expire messages for all subscription on topic {}",
- clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ future = CompletableFuture.completedFuture(null);
}
- // If the topic name is a partition name, no need to get partition topic metadata again
- if (topicName.isPartitioned()) {
- internalExpireMessagesForAllSubscriptionsForNonPartitionedTopic(asyncResponse,
+ future.thenCompose(__ -> {
+ // If the topic name is a partition name, no need to get partition topic metadata again
+ if (topicName.isPartitioned()) {
+ return internalExpireMessagesForAllSubscriptionsForNonPartitionedTopicAsync(asyncResponse,
expireTimeInSeconds, authoritative);
- } else {
- getPartitionedTopicMetadataAsync(topicName,
- authoritative, false).thenAccept(partitionMetadata -> {
- if (partitionMetadata.partitions > 0) {
- final List<CompletableFuture<Void>> futures = Lists.newArrayList();
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenCompose(partitionMetadata -> {
+ if (partitionMetadata.partitions > 0) {
+ final List<CompletableFuture<Void>> futures = Lists.newArrayList();
- // expire messages for each partition topic
- for (int i = 0; i < partitionMetadata.partitions; i++) {
- TopicName topicNamePartition = topicName.getPartition(i);
- try {
- futures.add(pulsar()
- .getAdminClient()
- .topics()
- .expireMessagesForAllSubscriptionsAsync(
+ // expire messages for each partition topic
+ for (int i = 0; i < partitionMetadata.partitions; i++) {
+ TopicName topicNamePartition = topicName.getPartition(i);
+ try {
+ futures.add(pulsar()
+ .getAdminClient()
+ .topics()
+ .expireMessagesForAllSubscriptionsAsync(
topicNamePartition.toString(), expireTimeInSeconds));
- } catch (Exception e) {
- log.error("[{}] Failed to expire messages up to {} on {}",
- clientAppId(), expireTimeInSeconds,
- topicNamePartition, e);
- asyncResponse.resume(new RestException(e));
- return;
- }
- }
+ } catch (Exception e) {
+ log.error("[{}] Failed to expire messages up to {} on {}",
+ clientAppId(), expireTimeInSeconds,
+ topicNamePartition, e);
+ asyncResponse.resume(new RestException(e));
+ return CompletableFuture.completedFuture(null);
+ }
+ }
- FutureUtil.waitForAll(futures).handle((result, exception) -> {
- if (exception != null) {
- Throwable t = exception.getCause();
- log.error("[{}] Failed to expire messages up to {} on {}",
- clientAppId(), expireTimeInSeconds,
- topicName, t);
- asyncResponse.resume(new RestException(t));
- return null;
+ return FutureUtil.waitForAll(futures).handle((result, exception) -> {
+ if (exception != null) {
+ Throwable t = exception.getCause();
+ log.error("[{}] Failed to expire messages up to {} on {}",
+ clientAppId(), expireTimeInSeconds,
+ topicName, t);
+ asyncResponse.resume(new RestException(t));
+ return null;
+ }
+ asyncResponse.resume(Response.noContent().build());
+ return null;
+ });
+ } else {
+ return internalExpireMessagesForAllSubscriptionsForNonPartitionedTopicAsync(asyncResponse,
+ expireTimeInSeconds, authoritative);
}
-
- asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(ex -> {
+ log.error("[{}] Failed to expire messages for all subscription on topic {}",
+ clientAppId(), topicName, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
return null;
});
- } else {
- internalExpireMessagesForAllSubscriptionsForNonPartitionedTopic(asyncResponse,
- expireTimeInSeconds, authoritative);
- }
- }).exceptionally(ex -> {
- log.error("[{}] Failed to expire messages for all subscription on topic {}",
- clientAppId(), topicName, ex);
- resumeAsyncResponseExceptionally(asyncResponse, ex);
- return null;
- });
- }
- }
-
- private void internalExpireMessagesForAllSubscriptionsForNonPartitionedTopic(AsyncResponse asyncResponse,
- int expireTimeInSeconds,
- boolean authoritative) {
- // validate ownership and redirect if current broker is not owner
- PersistentTopic topic;
- try {
- validateTopicOwnership(topicName, authoritative);
- validateTopicOperation(topicName, TopicOperation.EXPIRE_MESSAGES);
- topic = (PersistentTopic) getTopicReference(topicName);
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to expire messages for all subscription on topic {},"
- + " redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- return;
- } catch (Exception e) {
- log.error("[{}] Failed to expire messages for all subscription on topic {}",
- clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- final AtomicReference<Throwable> exception = new AtomicReference<>();
-
- topic.getReplicators().forEach((subName, replicator) -> {
- try {
- internalExpireMessagesByTimestampForSinglePartition(subName, expireTimeInSeconds, authoritative);
- } catch (Throwable t) {
- exception.set(t);
- }
- });
-
- topic.getSubscriptions().forEach((subName, subscriber) -> {
- try {
- internalExpireMessagesByTimestampForSinglePartition(subName, expireTimeInSeconds, authoritative);
- } catch (Throwable t) {
- exception.set(t);
}
});
+ }
- if (exception.get() != null) {
- if (exception.get() instanceof WebApplicationException) {
- WebApplicationException wae = (WebApplicationException) exception.get();
- asyncResponse.resume(wae);
- return;
- } else {
- asyncResponse.resume(new RestException(exception.get()));
- return;
- }
- }
+ private CompletableFuture<Void> internalExpireMessagesForAllSubscriptionsForNonPartitionedTopicAsync(
+ AsyncResponse asyncResponse,
+ int expireTimeInSeconds,
+ boolean authoritative) {
+ // validate ownership and redirect if current broker is not owner
+ return validateTopicOwnershipAsync(topicName, authoritative)
+ .thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.EXPIRE_MESSAGES))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(t -> {
+ PersistentTopic topic = (PersistentTopic) t;
+ final AtomicReference<Throwable> exception = new AtomicReference<>();
+
+ List<CompletableFuture<Void>> futures = Lists.newArrayList();
+ topic.getReplicators().forEach((subName, replicator) ->
+ futures.add(
+ internalExpireMessagesByTimestampForSinglePartitionAsync(subName, expireTimeInSeconds,
+ authoritative)
+ .exceptionally(ex -> {
+ exception.set(FutureUtil.unwrapCompletionException(ex));
+ return null;
+ })));
+
+ topic.getSubscriptions().forEach((subName, subscriber) ->
+ futures.add(
+ internalExpireMessagesByTimestampForSinglePartitionAsync(subName, expireTimeInSeconds,
+ authoritative)
+ .exceptionally(ex -> {
+ exception.set(FutureUtil.unwrapCompletionException(ex));
+ return null;
+ })));
- asyncResponse.resume(Response.noContent().build());
+ return FutureUtil.waitForAll(futures).thenRun(() -> {
Review comment:
We can use `exceptionally` here to handle the exceptions. No need for the `exception` in Line 1876
--
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