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 10:20:00 UTC
[GitHub] [pulsar] codelipenghui commented on a change in pull request #13853: [Broker] Fix call sync method in async rest api for internalTriggerCompaction
codelipenghui commented on a change in pull request #13853:
URL: https://github.com/apache/pulsar/pull/13853#discussion_r789532889
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -3543,113 +3543,104 @@ protected void internalExpireMessagesByPosition(AsyncResponse asyncResponse, Str
protected void internalTriggerCompaction(AsyncResponse asyncResponse, boolean authoritative) {
log.info("[{}] Trigger compaction on topic {}", clientAppId(), topicName);
- try {
- if (topicName.isGlobal()) {
- validateGlobalNamespaceOwnership(namespaceName);
- }
- } catch (Exception e) {
- if (e instanceof WebApplicationException) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to trigger compaction on topic {}, redirecting to other brokers.",
- clientAppId(), topicName, e);
- }
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- log.error("[{}] Failed to trigger compaction on topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
-
- // If the topic name is a partition name, no need to get partition topic metadata again
- if (topicName.isPartitioned()) {
- try {
- internalTriggerCompactionNonPartitionedTopic(authoritative);
- } catch (Exception e) {
- log.error("[{}] Failed to trigger compaction on topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- asyncResponse.resume(Response.noContent().build());
+ CompletableFuture<Void> future;
+ if (topicName.isGlobal()) {
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
} else {
- getPartitionedTopicMetadataAsync(topicName, authoritative, false).thenAccept(partitionMetadata -> {
- final int numPartitions = partitionMetadata.partitions;
- if (numPartitions > 0) {
- final List<CompletableFuture<Void>> futures = Lists.newArrayList();
-
- for (int i = 0; i < numPartitions; i++) {
- TopicName topicNamePartition = topicName.getPartition(i);
- try {
- futures.add(pulsar()
- .getAdminClient()
- .topics()
- .triggerCompactionAsync(topicNamePartition.toString()));
- } catch (Exception e) {
- log.error("[{}] Failed to trigger compaction on topic {}",
- clientAppId(), topicNamePartition, e);
- asyncResponse.resume(new RestException(e));
- return;
- }
- }
+ future = CompletableFuture.completedFuture(null);
+ }
+ future.thenAccept(__ -> {
+ // If the topic name is a partition name, no need to get partition topic metadata again
+ if (topicName.isPartitioned()) {
+ internalTriggerCompactionNonPartitionedTopic(asyncResponse, authoritative);
+ } else {
+ getPartitionedTopicMetadataAsync(topicName, authoritative, false).thenAccept(partitionMetadata -> {
+ final int numPartitions = partitionMetadata.partitions;
+ if (numPartitions > 0) {
+ final List<CompletableFuture<Void>> futures = Lists.newArrayListWithCapacity(numPartitions);
- FutureUtil.waitForAll(futures).handle((result, exception) -> {
- if (exception != null) {
- Throwable th = exception.getCause();
- if (th instanceof NotFoundException) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND, th.getMessage()));
- return null;
- } else if (th instanceof WebApplicationException) {
- asyncResponse.resume(th);
- return null;
- } else {
+ for (int i = 0; i < numPartitions; i++) {
+ TopicName topicNamePartition = topicName.getPartition(i);
+ try {
+ futures.add(pulsar()
+ .getAdminClient()
+ .topics()
+ .triggerCompactionAsync(topicNamePartition.toString()));
+ } catch (Exception e) {
log.error("[{}] Failed to trigger compaction on topic {}",
- clientAppId(), topicName, exception);
- asyncResponse.resume(new RestException(exception));
- return null;
+ clientAppId(), topicNamePartition, e);
+ asyncResponse.resume(new RestException(e));
+ return;
}
}
- asyncResponse.resume(Response.noContent().build());
- return null;
- });
- } else {
+
+ FutureUtil.waitForAll(futures).handle((result, exception) -> {
+ if (exception != null) {
+ Throwable th = exception.getCause();
+ if (th instanceof NotFoundException) {
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, th.getMessage()));
+ return null;
+ } else if (th instanceof WebApplicationException) {
+ asyncResponse.resume(th);
+ return null;
+ } else {
+ log.error("[{}] Failed to trigger compaction on topic {}",
+ clientAppId(), topicName, exception);
+ asyncResponse.resume(new RestException(exception));
+ return null;
+ }
+ }
+ asyncResponse.resume(Response.noContent().build());
+ return null;
+ });
+ } else {
+ internalTriggerCompactionNonPartitionedTopic(asyncResponse, authoritative);
+ }
+ }).exceptionally(ex -> {
+ Throwable cause = ex.getCause();
+ log.error("[{}] Failed to validate global namespace ownership to trigger compaction on topic {}",
+ clientAppId(), topicName, cause);
+ resumeAsyncResponseExceptionally(asyncResponse, cause);
+ return null;
+ });
+ }
+ });
+ }
+
+ protected void internalTriggerCompactionNonPartitionedTopic(AsyncResponse asyncResponse, boolean authoritative) {
+ validateTopicOwnershipAsync(topicName, authoritative)
+ .thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.COMPACT))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenAccept(topic -> {
try {
- internalTriggerCompactionNonPartitionedTopic(authoritative);
+ ((PersistentTopic) topic).triggerCompaction();
+ asyncResponse.resume(Response.noContent().build());
+ } catch (AlreadyRunningException e) {
+ resumeAsyncResponseExceptionally(asyncResponse,
+ new RestException(Status.CONFLICT, e.getMessage()));
+ return;
} catch (Exception e) {
- if (e instanceof WebApplicationException) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to trigger compaction on topic {}, "
- + "redirecting to other brokers.", clientAppId(), topicName, e);
- }
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- log.error("[{}] Failed to trigger compaction on topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
+ log.error("[{}] Failed to trigger compaction on topic {}", clientAppId(),
+ topicName, e);
+ resumeAsyncResponseExceptionally(asyncResponse, new RestException(new RestException(e)));
Review comment:
```suggestion
resumeAsyncResponseExceptionally(asyncResponse, new RestException(e));
```
--
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