You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2021/12/21 02:18:46 UTC
[pulsar] 14/16: Fix when deleting topic with NotFoundException, do not return to client (#13203)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9daba91d8f66627963ada9041699cfee496e8d20
Author: Jiwei Guo <te...@apache.org>
AuthorDate: Sat Dec 11 15:20:49 2021 +0800
Fix when deleting topic with NotFoundException, do not return to client (#13203)
(cherry picked from commit bd68b6f05f9749328701c59bdaf3cddda2254d39)
---
.../apache/pulsar/broker/admin/AdminResource.java | 8 ++++++
.../broker/admin/impl/PersistentTopicsBase.java | 5 ++--
.../pulsar/broker/admin/PersistentTopicsTest.java | 32 ++++++++++++++++++++++
3 files changed, 42 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
index f3a94d2..9045462 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
@@ -35,6 +35,7 @@ import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.systopic.SystemTopicClient;
@@ -61,6 +62,7 @@ import org.apache.pulsar.common.policies.data.impl.DispatchRateImpl;
import org.apache.pulsar.common.util.Codec;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.ObjectMapperFactory;
+import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.MetadataStoreException.AlreadyExistsException;
import org.apache.pulsar.metadata.api.MetadataStoreException.BadVersionException;
@@ -772,6 +774,12 @@ public abstract class AdminResource extends PulsarWebResource {
}
}
+ protected boolean isManagedLedgerNotFoundException(Exception e) {
+ Throwable cause = e.getCause();
+ return cause instanceof ManagedLedgerException.MetadataNotFoundException
+ || cause instanceof MetadataStoreException.NotFoundException;
+ }
+
protected void checkArgument(boolean b, String errorMessage) {
if (!b) {
throw new RestException(Status.BAD_REQUEST, errorMessage);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index be76ff1..5a084bf 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -56,7 +56,6 @@ import org.apache.bookkeeper.mledger.LedgerOffloader;
import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
-import org.apache.bookkeeper.mledger.ManagedLedgerException.MetadataNotFoundException;
import org.apache.bookkeeper.mledger.ManagedLedgerInfo;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
@@ -302,7 +301,7 @@ public class PersistentTopicsBase extends AdminResource {
try {
pulsar().getBrokerService().deleteTopic(topicName.toString(), true, deleteSchema).get();
} catch (Exception e) {
- if (e.getCause() instanceof MetadataNotFoundException) {
+ if (isManagedLedgerNotFoundException(e)) {
log.info("[{}] Topic was already not existing {}", clientAppId(), topicName, e);
} else {
log.error("[{}] Failed to delete topic forcefully {}", clientAppId(), topicName, e);
@@ -1021,7 +1020,7 @@ public class PersistentTopicsBase extends AdminResource {
log.error("[{}] Failed to delete topic {}", clientAppId(), topicName, t);
if (t instanceof TopicBusyException) {
throw new RestException(Status.PRECONDITION_FAILED, "Topic has active producers/subscriptions");
- } else if (t instanceof MetadataNotFoundException) {
+ } else if (isManagedLedgerNotFoundException(e)) {
throw new RestException(Status.NOT_FOUND, "Topic not found");
} else {
throw new RestException(t);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
index 6e5ef99..e41db38 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
@@ -54,6 +54,7 @@ import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.broker.authentication.AuthenticationDataHttps;
import org.apache.pulsar.broker.resources.PulsarResources;
import org.apache.pulsar.broker.resources.TopicResources;
+import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.web.PulsarWebResource;
import org.apache.pulsar.broker.web.RestException;
import org.apache.pulsar.client.admin.PulsarAdminException;
@@ -78,6 +79,7 @@ import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.policies.data.TopicStats;
+import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.zookeeper.KeeperException;
import org.mockito.ArgumentCaptor;
import org.powermock.reflect.Whitebox;
@@ -1084,4 +1086,34 @@ public class PersistentTopicsTest extends MockedPulsarServiceBaseTest {
Assert.assertTrue(admin.topics().getMessageIdByTimestamp(topicName, publish2 + 1)
.compareTo(id2) > 0);
}
+
+ @Test
+ public void testDeleteTopic() throws Exception {
+ final String topicName = "topic-1";
+ BrokerService brokerService = spy(pulsar.getBrokerService());
+ doReturn(brokerService).when(pulsar).getBrokerService();
+ persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, false);
+ CompletableFuture<Void> deleteTopicFuture = new CompletableFuture<>();
+ deleteTopicFuture.completeExceptionally(new MetadataStoreException.NotFoundException());
+ doReturn(deleteTopicFuture).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean());
+ persistentTopics.deleteTopic(testTenant, testNamespace, topicName, true, true, true);
+ //
+ CompletableFuture<Void> deleteTopicFuture2 = new CompletableFuture<>();
+ deleteTopicFuture2.completeExceptionally(new MetadataStoreException("test exception"));
+ doReturn(deleteTopicFuture2).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean());
+ try {
+ persistentTopics.deleteTopic(testTenant, testNamespace, topicName, true, true, true);
+ } catch (Exception e) {
+ Assert.assertTrue(e instanceof RestException);
+ }
+ //
+ CompletableFuture<Void> deleteTopicFuture3 = new CompletableFuture<>();
+ deleteTopicFuture3.completeExceptionally(new MetadataStoreException.NotFoundException());
+ doReturn(deleteTopicFuture3).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean());
+ try {
+ persistentTopics.deleteTopic(testTenant, testNamespace, topicName, false, true, true);
+ } catch (RestException e) {
+ Assert.assertEquals(e.getResponse().getStatus(), 404);
+ }
+ }
}