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 2021/04/25 08:21:58 UTC
[GitHub] [pulsar] gaoran10 commented on a change in pull request #10326: support truncate topic
gaoran10 commented on a change in pull request #10326:
URL: https://github.com/apache/pulsar/pull/10326#discussion_r619753621
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerFactory.java
##########
@@ -18,6 +18,7 @@
*/
package org.apache.bookkeeper.mledger;
+import java.util.concurrent.CompletableFuture;
Review comment:
Is this import necessary?
##########
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:
It seems that this block repeated with above.
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
##########
@@ -600,4 +600,10 @@ void asyncSetProperties(Map<String, String> properties, final AsyncCallbacks.Upd
* will got null if corresponding ledger not exists.
*/
CompletableFuture<LedgerInfo> getLedgerInfo(long ledgerId);
+
+ /**
+ * Truncate ledgers
+ * The latest ledger cannot be deleted ,and only delete acknowledged ledgers
Review comment:
Only delete ledgers before the mark delete position, right?
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -3738,6 +3751,35 @@ public void setEntriesAddedCounter(long count) {
private static final Logger log = LoggerFactory.getLogger(ManagedLedgerImpl.class);
+ @Override
+ public CompletableFuture<Void> asyncTruncate() {
+
+ final List<CompletableFuture<Void>> futures = Lists.newArrayList();
+ for (ManagedCursor cursor : cursors) {
+ final CompletableFuture<Void> future = new CompletableFuture<>();
+ cursor.asyncClearBacklog(new AsyncCallbacks.ClearBacklogCallback() {
Review comment:
Why need to clear backlog?
##########
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:
It seems that this block is repeated.
--
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:
users@infra.apache.org