jangwind commented on a change in pull request #10326:
URL: https://github.com/apache/pulsar/pull/10326#discussion_r619809608
##########
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:
Sorry, I added this dependency in the previous modification, I will
delete this dependency.
##########
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.
##########
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:
If it doesn't clearly backLog, it will be trim instead of truncate. So I
do it.
##########
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:
You are right. The description has been changed to ‘The truncate
operation will move all cursors to the end of the topic and delete all inactive
ledgers.’
--
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]
