[GitHub] [pulsar] 315157973 commented on a diff in pull request #19094: [feat][admin] PIP-219 Part-1 Add admin API for trimming topic

[GitBox <gi...@apache.org>]

315157973 commented on code in PR #19094:
URL: https://github.com/apache/pulsar/pull/19094#discussion_r1059049287


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -4655,6 +4656,74 @@ protected void internalGetLastMessageId(AsyncResponse asyncResponse, boolean aut
                     return null;
         });
     }
+    protected void internalTrimTopic(AsyncResponse asyncResponse, boolean authoritative) {
+        validateTopicOwnershipAsync(topicName, authoritative)
+                .thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.TRIM_TOPIC))
+                .thenCompose(__ -> getTopicReferenceAsync(topicName))
+                .thenCompose(topic -> {
+                    if (topic == null) {
+                        asyncResponse.resume(new RestException(Status.NOT_FOUND,
+                                getTopicNotFoundErrorMessage(topicName.toString())));
+                        return null;
+                    }
+                    if (!(topic instanceof PersistentTopic)) {
+                        log.info("[{}] Trim on a non-persistent topic {} is not allowed", clientAppId(), topicName);
+                        asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED,
+                                "Trim on a non-persistent topic is not allowed"));
+                        return null;
+                    }
+                    if (topicName.isPartitioned()) {
+                        return trimPartitionedTopic(asyncResponse, authoritative);
+                    }
+                    return trimNonPartitionedTopic(asyncResponse, (PersistentTopic) topic);
+                }).exceptionally(ex -> {
+                    // If the exception is not redirect exception we need to log it.
+                    if (!isRedirectException(ex)) {
+                        log.error("[{}] Failed to trim topic {}", clientAppId(), topicName, ex);
+                    }
+                    resumeAsyncResponseExceptionally(asyncResponse, ex);
+                    return null;
+                });
+    }
+
+    @NotNull
+    private static CompletableFuture<Void> trimNonPartitionedTopic(AsyncResponse asyncResponse,
+                                                                   PersistentTopic topic) {
+        CompletableFuture<Void> result = new CompletableFuture<>();
+        Optional.ofNullable(topic.getManagedLedger()).ifPresent(managedLedger ->

Review Comment:
   Good catch



-- 
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