jangwind commented on a change in pull request #10326:
URL: https://github.com/apache/pulsar/pull/10326#discussion_r619815086
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -3922,4 +3922,89 @@ protected void internalHandleResult(AsyncResponse
asyncResponse,
}
}
}
+
+ protected void internalTruncateTopic(AsyncResponse asyncResponse, boolean
authoritative) {
+
+ // If the topic name is a partition name, no need to get partition
topic metadata again
+ if (topicName.isPartitioned()) {
+ Topic topic;
+ try {
+ validateAdminAccessForTenant(topicName.getTenant());
+ validateTopicOwnership(topicName, authoritative);
+ topic = getTopicReference(topicName);
+ } catch (Exception e) {
+ log.error("[{}] Failed to truncate topic {}", clientAppId(),
topicName, e);
+ resumeAsyncResponseExceptionally(asyncResponse, e);
+ return;
+ }
+ CompletableFuture<Void> future = topic.truncate();
+ future.thenAccept(a -> {
+ asyncResponse.resume(new
RestException(Response.Status.NO_CONTENT.getStatusCode(),
+ Response.Status.NO_CONTENT.getReasonPhrase()));
+ }).exceptionally(e -> {
+ asyncResponse.resume(e);
+ return null;
+ });
Review comment:
Thank you for your comment. This block is based on
(topicName.isPartitioned()==true) and another is based on (meta.partitions ==
0). I will merge them Into a function.
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -3922,4 +3922,89 @@ protected void internalHandleResult(AsyncResponse
asyncResponse,
}
}
}
+
+ protected void internalTruncateTopic(AsyncResponse asyncResponse, boolean
authoritative) {
+
+ // If the topic name is a partition name, no need to get partition
topic metadata again
+ if (topicName.isPartitioned()) {
+ Topic topic;
+ try {
+ validateAdminAccessForTenant(topicName.getTenant());
+ validateTopicOwnership(topicName, authoritative);
+ topic = getTopicReference(topicName);
+ } catch (Exception e) {
+ log.error("[{}] Failed to truncate topic {}", clientAppId(),
topicName, e);
+ resumeAsyncResponseExceptionally(asyncResponse, e);
+ return;
+ }
+ CompletableFuture<Void> future = topic.truncate();
+ future.thenAccept(a -> {
+ asyncResponse.resume(new
RestException(Response.Status.NO_CONTENT.getStatusCode(),
+ Response.Status.NO_CONTENT.getReasonPhrase()));
+ }).exceptionally(e -> {
+ asyncResponse.resume(e);
+ return null;
+ });
+ } else {
+ getPartitionedTopicMetadataAsync(topicName, authoritative,
false).whenComplete((meta, t) -> {
+ if (meta.partitions > 0) {
+ final List<CompletableFuture<Void>> futures =
Lists.newArrayList();
+ for (int i = 0; i < meta.partitions; i++) {
+ TopicName topicNamePartition =
topicName.getPartition(i);
+ try {
+ futures.add(pulsar().getAdminClient().topics()
+ .truncateAsync(topicNamePartition.toString()));
+ } catch (Exception e) {
+ log.error("[{}] Failed to truncate topic {}",
clientAppId(), topicNamePartition, e);
+ asyncResponse.resume(new RestException(e));
+ return;
+ }
+ }
+ 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()));
+ } else if (th instanceof WebApplicationException) {
+ asyncResponse.resume(th);
+ } else {
+ log.error("[{}] Failed to truncate topic {}",
clientAppId(), topicName, exception);
+ asyncResponse.resume(new
RestException(exception));
+ }
+ } else {
+ asyncResponse.resume(Response.noContent().build());
+ }
+ return null;
+ });
+ } else {
+ Topic topic;
+ try {
+ validateAdminAccessForTenant(topicName.getTenant());
+ validateTopicOwnership(topicName, authoritative);
+ topic = getTopicReference(topicName);
+ } catch (Exception e) {
+ log.error("[{}] Failed to truncate topic {}",
clientAppId(), topicName, e);
+ resumeAsyncResponseExceptionally(asyncResponse, e);
+ return;
+ }
+ CompletableFuture<Void> future = topic.truncate();
+ future.thenAccept(a -> {
+ asyncResponse.resume(new
RestException(Response.Status.NO_CONTENT.getStatusCode(),
+ Response.Status.NO_CONTENT.getReasonPhrase()));
+ }).exceptionally(e -> {
+ asyncResponse.resume(e);
+ return null;
+ });
Review comment:
Thank you for your comment. This block is based on
(topicName.isPartitioned()==true) and another is based on (meta.partitions ==
0). I will merge them Into a function.
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
