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 03:50:19 UTC
[GitHub] [pulsar] AnonHxy opened a new pull request #13905: [WIP][Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…
AnonHxy opened a new pull request #13905:
URL: https://github.com/apache/pulsar/pull/13905
Master Issue: #<https://github.com/apache/pulsar/issues/13854>
### Motivation
See Issue: #<https://github.com/apache/pulsar/issues/13854>
### Modifications
### Verifying this change
- [x] Make sure that the change passes the CI checks.
### Does this pull request potentially affect one of the following parts:
*If `yes` was chosen, please highlight the changes*
- Dependencies (does it add or upgrade a dependency): (no)
- The public API: (no)
- The schema: (no)
- The default values of configurations: (no)
- The wire protocol: (no)
- The rest endpoints: (no)
- The admin cli options: (no)
- Anything that affects deployment: (no)
### Documentation
- [x] `no-need-doc`
--
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
[GitHub] [pulsar] liudezhi2098 commented on a change in pull request #13905: [Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…
Posted by GitBox <gi...@apache.org>.
liudezhi2098 commented on a change in pull request #13905:
URL: https://github.com/apache/pulsar/pull/13905#discussion_r790262801
##########
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:
conflict with #13880
--
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
[GitHub] [pulsar] AnonHxy commented on pull request #13905: [Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on pull request #13905:
URL: https://github.com/apache/pulsar/pull/13905#issuecomment-1020877964
/pulsarbot run-failure-checks
--
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
[GitHub] [pulsar] AnonHxy commented on a change in pull request #13905: [Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on a change in pull request #13905:
URL: https://github.com/apache/pulsar/pull/13905#discussion_r790756168
##########
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:
update
--
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
[GitHub] [pulsar] AnonHxy commented on a change in pull request #13905: [Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on a change in pull request #13905:
URL: https://github.com/apache/pulsar/pull/13905#discussion_r790268283
##########
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:
This `exceptionally` can handle `getPartitionedTopicMetadataAsync` exceptions if `topicName.isPartitioned() ` return false
--
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
[GitHub] [pulsar] AnonHxy commented on a change in pull request #13905: [Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on a change in pull request #13905:
URL: https://github.com/apache/pulsar/pull/13905#discussion_r790757687
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -1801,124 +1801,129 @@ 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 -> {
+ Throwable cause = FutureUtil.unwrapCompletionException(ex);
+ log.error("[{}] Failed to expire messages for all subscription on topic {}",
+ clientAppId(), topicName, cause);
+ resumeAsyncResponseExceptionally(asyncResponse, cause);
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<>();
Review comment:
looks the same as https://github.com/apache/pulsar/pull/13905#discussion_r790265218? I will update there code
--
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
[GitHub] [pulsar] AnonHxy closed pull request #13905: [Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…
Posted by GitBox <gi...@apache.org>.
AnonHxy closed pull request #13905:
URL: https://github.com/apache/pulsar/pull/13905
--
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
[GitHub] [pulsar] AnonHxy commented on pull request #13905: [Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on pull request #13905:
URL: https://github.com/apache/pulsar/pull/13905#issuecomment-1020877964
/pulsarbot run-failure-checks
--
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
[GitHub] [pulsar] AnonHxy commented on a change in pull request #13905: [Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on a change in pull request #13905:
URL: https://github.com/apache/pulsar/pull/13905#discussion_r790270597
##########
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:
Yes, there miss return
--
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
[GitHub] [pulsar] AnonHxy commented on a change in pull request #13905: [Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on a change in pull request #13905:
URL: https://github.com/apache/pulsar/pull/13905#discussion_r790603658
##########
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:
Ok, I can rebase the code after https://github.com/apache/pulsar/pull/13880 merging to master to fix the conflict
--
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
[GitHub] [pulsar] AnonHxy commented on a change in pull request #13905: [Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on a change in pull request #13905:
URL: https://github.com/apache/pulsar/pull/13905#discussion_r790265700
##########
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:
comment in https://github.com/apache/pulsar/pull/13880
--
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
[GitHub] [pulsar] AnonHxy commented on a change in pull request #13905: [Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on a change in pull request #13905:
URL: https://github.com/apache/pulsar/pull/13905#discussion_r790271240
##########
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:
If `(!topicName.isPartitioned()) && metadata.partitions == 0` , we should also `doInternalExpireMessagesByTimestampForSinglePartitionAsync`
--
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
[GitHub] [pulsar] codelipenghui commented on a change in pull request #13905: [Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #13905:
URL: https://github.com/apache/pulsar/pull/13905#discussion_r790507438
##########
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:
Oh, I think we'd better close the new one and apply the fix from the old one.
##########
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:
+1
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -1801,124 +1801,129 @@ 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 -> {
+ Throwable cause = FutureUtil.unwrapCompletionException(ex);
+ log.error("[{}] Failed to expire messages for all subscription on topic {}",
+ clientAppId(), topicName, cause);
+ resumeAsyncResponseExceptionally(asyncResponse, cause);
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<>();
Review comment:
Why do we need this line? You can get the exception from the `futures` from line 1879 directly.
--
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
[GitHub] [pulsar] AnonHxy commented on a change in pull request #13905: [Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on a change in pull request #13905:
URL: https://github.com/apache/pulsar/pull/13905#discussion_r790265700
##########
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:
comment in <https://github.com/apache/pulsar/pull/13880/files>
--
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
[GitHub] [pulsar] Jason918 commented on a change in pull request #13905: [Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…
Posted by GitBox <gi...@apache.org>.
Jason918 commented on a change in pull request #13905:
URL: https://github.com/apache/pulsar/pull/13905#discussion_r790370822
##########
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:
> This `exceptionally` can handle `getPartitionedTopicMetadataAsync` exceptions if `topicName.isPartitioned() ` return false
Yes, but exceptionally is also needed there for `validateGlobalNamespaceOwnership`
--
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
[GitHub] [pulsar] AnonHxy commented on a change in pull request #13905: [Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on a change in pull request #13905:
URL: https://github.com/apache/pulsar/pull/13905#discussion_r790755891
##########
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:
update
--
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
[GitHub] [pulsar] Jason918 commented on a change in pull request #13905: [Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…
Posted by GitBox <gi...@apache.org>.
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
[GitHub] [pulsar] liudezhi2098 commented on a change in pull request #13905: [Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…
Posted by GitBox <gi...@apache.org>.
liudezhi2098 commented on a change in pull request #13905:
URL: https://github.com/apache/pulsar/pull/13905#discussion_r790268510
##########
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:
Both PR's modify this method，warning: There may be conflict.
--
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