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:
[email protected]
