diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java index c2ce36d49ffe7..f76e8a02827ab 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java @@ -73,6 +73,7 @@ import org.apache.pulsar.common.naming.NamespaceBundleSplitAlgorithm; import org.apache.pulsar.common.naming.NamespaceBundles; import org.apache.pulsar.common.naming.NamespaceName; +import org.apache.pulsar.common.naming.SystemTopicNames; import org.apache.pulsar.common.naming.TopicDomain; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.policies.data.AuthAction; @@ -310,10 +311,21 @@ protected void internalDeleteNamespace(AsyncResponse asyncResponse, boolean auth // remove from owned namespace map and ephemeral node from ZK final List> futures = Lists.newArrayList(); // remove system topics first. + Set noPartitionedTopicPolicySystemTopic = new HashSet<>(); + Set partitionedTopicPolicySystemTopic = new HashSet<>(); if (!topics.isEmpty()) { for (String topic : topics) { try { - futures.add(pulsar().getAdminClient().topics().deleteAsync(topic, true, true)); + if (SystemTopicNames.isTopicPoliciesSystemTopic(topic)) { + TopicName topicName = TopicName.get(topic); + if (topicName.isPartitioned()) { + partitionedTopicPolicySystemTopic.add(topicName.getPartitionedTopicName()); + } else { + noPartitionedTopicPolicySystemTopic.add(topic); + } + } else { + futures.add(pulsar().getAdminClient().topics().deleteAsync(topic, true, true)); + } } catch (Exception ex) { log.error("[{}] Failed to delete system topic {}", clientAppId(), topic, ex); asyncResponse.resume(new RestException(Status.INTERNAL_SERVER_ERROR, ex)); @@ -321,11 +333,14 @@ protected void internalDeleteNamespace(AsyncResponse asyncResponse, boolean auth } } } - FutureUtil.waitForAll(futures).thenCompose(__ -> { - List> deleteBundleFutures = Lists.newArrayList(); - NamespaceBundles bundles = pulsar().getNamespaceService().getNamespaceBundleFactory() - .getBundles(namespaceName); - for (NamespaceBundle bundle : bundles.getBundles()) { + FutureUtil.waitForAll(futures) + .thenCompose(ignore -> internalDeleteTopicsAsync(noPartitionedTopicPolicySystemTopic)) + .thenCompose(ignore -> internalDeletePartitionedTopicsAsync(partitionedTopicPolicySystemTopic)) + .thenCompose(__ -> { + List> deleteBundleFutures = Lists.newArrayList(); + NamespaceBundles bundles = pulsar().getNamespaceService().getNamespaceBundleFactory() + .getBundles(namespaceName); + for (NamespaceBundle bundle : bundles.getBundles()) { // check if the bundle is owned by any broker, if not then we do not need to delete the bundle deleteBundleFutures.add(pulsar().getNamespaceService().getOwnerAsync(bundle).thenCompose(ownership -> { if (ownership.isPresent()) { @@ -475,13 +490,19 @@ protected void internalDeleteNamespaceForcefully(AsyncResponse asyncResponse, bo Set nonPartitionedTopics = new HashSet<>(); Set allSystemTopics = new HashSet<>(); Set allPartitionedSystemTopics = new HashSet<>(); + Set noPartitionedTopicPolicySystemTopic = new HashSet<>(); + Set partitionedTopicPolicySystemTopic = new HashSet<>(); for (String topic : topics) { try { TopicName topicName = TopicName.get(topic); if (topicName.isPartitioned()) { if (pulsar().getBrokerService().isSystemTopic(topicName)) { - allPartitionedSystemTopics.add(topicName.getPartitionedTopicName()); + if (SystemTopicNames.isTopicPoliciesSystemTopic(topic)) { + partitionedTopicPolicySystemTopic.add(topicName.getPartitionedTopicName()); + } else { + allPartitionedSystemTopics.add(topicName.getPartitionedTopicName()); + } continue; } String partitionedTopic = topicName.getPartitionedTopicName(); @@ -490,12 +511,17 @@ protected void internalDeleteNamespaceForcefully(AsyncResponse asyncResponse, bo } } else { if (pulsar().getBrokerService().isSystemTopic(topicName)) { - allSystemTopics.add(topic); + if (SystemTopicNames.isTopicPoliciesSystemTopic(topic)) { + noPartitionedTopicPolicySystemTopic.add(topic); + } else { + allSystemTopics.add(topic); + } continue; } nonPartitionedTopics.add(topic); } - topicFutures.add(pulsar().getAdminClient().topics().deleteAsync(topic, true)); + topicFutures.add(pulsar().getAdminClient().topics().deleteAsync( + topic, true, true)); } catch (Exception e) { String errorMessage = String.format("Failed to force delete topic %s, " + "but the previous deletion command of partitioned-topics:%s " @@ -524,6 +550,9 @@ protected void internalDeleteNamespaceForcefully(AsyncResponse asyncResponse, bo .thenCompose((ignore) -> internalDeleteTopicsAsync(allSystemTopics)) .thenCompose((ignore) -> internalDeletePartitionedTopicsAsync(allPartitionedSystemTopics)) + .thenCompose(ignore -> + internalDeletePartitionedTopicsAsync(partitionedTopicPolicySystemTopic)) + .thenCompose(ignore -> internalDeleteTopicsAsync(noPartitionedTopicPolicySystemTopic)) .handle((result, exception) -> { if (exception != null) { if (exception.getCause() instanceof PulsarAdminException) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index 8e95f2c431361..d8f56d7296007 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -1186,7 +1186,7 @@ private CompletableFuture delete(boolean failIfHasSubscriptions, deleteTopicAuthenticationFuture.thenCompose(ignore -> deleteSchema()) .thenCompose(ignore -> { if (!SystemTopicNames.isTopicPoliciesSystemTopic(topic) - && brokerService.getPulsar().getConfiguration().isSystemTopicEnabled()) { + && brokerService.getPulsar().getConfiguration().isSystemTopicEnabled()) { return deleteTopicPolicies(); } else { return CompletableFuture.completedFuture(null); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java index 16dfe5bc9a3b6..eaa54c1fd3402 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.broker.admin; +import static org.apache.pulsar.common.naming.NamespaceName.SYSTEM_NAMESPACE; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doNothing; import static org.mockito.Mockito.doReturn; @@ -57,11 +58,13 @@ import javax.ws.rs.core.Response.Status; import javax.ws.rs.core.UriBuilder; import javax.ws.rs.core.UriInfo; +import lombok.Cleanup; import org.apache.bookkeeper.client.api.ReadHandle; import org.apache.bookkeeper.mledger.LedgerOffloader; import org.apache.bookkeeper.mledger.ManagedLedgerConfig; import org.apache.bookkeeper.util.ZkUtils; import org.apache.pulsar.broker.BrokerTestUtil; +import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.admin.v1.Namespaces; import org.apache.pulsar.broker.admin.v1.PersistentTopics; import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; @@ -81,6 +84,7 @@ import org.apache.pulsar.client.api.ConsumerBuilder; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.common.api.proto.CommandSubscribe; import org.apache.pulsar.common.naming.NamespaceBundle; @@ -103,6 +107,7 @@ import org.apache.pulsar.common.policies.data.SubscribeRate; import org.apache.pulsar.common.policies.data.TenantInfo; import org.apache.pulsar.common.policies.data.TenantInfoImpl; +import org.apache.pulsar.common.policies.data.TopicPolicies; import org.apache.pulsar.common.policies.data.impl.DispatchRateImpl; import org.apache.pulsar.metadata.cache.impl.MetadataCacheImpl; import org.apache.pulsar.metadata.impl.AbstractMetadataStore; @@ -1938,7 +1943,7 @@ public void testFinallyDeleteSystemTopicWhenDeleteNamespace() throws Exception { } @Test - public void testNotClearTopicPolicesWhenDeleteSystemTopic() throws Exception { + public void testNotClearTopicPolicesWhenDeleteTopicPolicyTopic() throws Exception { String namespace = this.testTenant + "/delete-systemTopic"; String topic = TopicName.get(TopicDomain.persistent.toString(), this.testTenant, "delete-systemTopic", "testNotClearTopicPolicesWhenDeleteSystemTopic").toString(); @@ -1958,4 +1963,33 @@ public void testNotClearTopicPolicesWhenDeleteSystemTopic() throws Exception { // 4. delete the policies topic and the topic wil not to clear topic polices admin.topics().delete(namespace + "/" + SystemTopicNames.NAMESPACE_EVENTS_LOCAL_NAME, true); } + @Test + public void testDeleteTopicPolicyWhenDeleteSystemTopic() throws Exception { + conf.setTopicLevelPoliciesEnabled(true); + conf.setSystemTopicEnabled(true); + Field field = PulsarService.class.getDeclaredField("topicPoliciesService"); + field.setAccessible(true); + field.set(pulsar, new SystemTopicBasedTopicPoliciesService(pulsar)); + + String systemTopic = SYSTEM_NAMESPACE.toString() + "/" + "testDeleteTopicPolicyWhenDeleteSystemTopic"; + admin.tenants().createTenant(SYSTEM_NAMESPACE.getTenant(), + new TenantInfoImpl(Set.of("role1", "role2"), Set.of("use", "usc", "usw"))); + + admin.namespaces().createNamespace(SYSTEM_NAMESPACE.toString()); + @Cleanup + Producer producer = pulsarClient.newProducer(Schema.STRING) + .topic(systemTopic).create(); + admin.topicPolicies().setMaxConsumers(systemTopic, 5); + + Integer maxConsumerPerTopic = pulsar + .getTopicPoliciesService() + .getTopicPoliciesBypassCacheAsync(TopicName.get(systemTopic)).get() + .getMaxConsumerPerTopic(); + + assertEquals(maxConsumerPerTopic, Integer.valueOf(5)); + admin.topics().delete(systemTopic, true); + TopicPolicies topicPolicies = pulsar.getTopicPoliciesService() + .getTopicPoliciesBypassCacheAsync(TopicName.get(systemTopic)).get(5, TimeUnit.SECONDS); + assertNull(topicPolicies); + } }