diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java index c5f3d508a56db..bcec8351733a9 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java @@ -3285,16 +3285,8 @@ public CompletableFuture deleteTopicPolicies(TopicName topicName) { if (!pulsarService.getConfig().isTopicLevelPoliciesEnabled()) { return CompletableFuture.completedFuture(null); } - return pulsarService.getPulsarResources().getNamespaceResources() - .getPoliciesAsync(topicName.getNamespaceObject()) - .thenCompose(optPolicies -> { - if (optPolicies.isPresent() && optPolicies.get().deleted) { - // We can return the completed future directly if the namespace is already deleted. - return CompletableFuture.completedFuture(null); - } - TopicName cloneTopicName = TopicName.get(topicName.getPartitionedTopicName()); - return pulsar.getTopicPoliciesService().deleteTopicPoliciesAsync(cloneTopicName); - }); + return pulsar.getTopicPoliciesService() + .deleteTopicPoliciesAsync(TopicName.get(topicName.getPartitionedTopicName())); } private CompletableFuture checkMaxTopicsPerNamespace(TopicName topicName, int numPartitions) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java index a24b1eeee101a..93a2cddad4754 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java @@ -105,50 +105,53 @@ private CompletableFuture sendTopicPolicyEvent(TopicName topicName, Action return CompletableFuture.failedFuture( new BrokerServiceException.NotAllowedException("Not allowed to send event to health check topic")); } - CompletableFuture result = new CompletableFuture<>(); - try { - createSystemTopicFactoryIfNeeded(); - } catch (PulsarServerException e) { - result.completeExceptionally(e); - return result; - } + return pulsarService.getPulsarResources().getNamespaceResources() + .getPoliciesAsync(topicName.getNamespaceObject()) + .thenCompose(namespacePolicies -> { + if (namespacePolicies.isPresent() && namespacePolicies.get().deleted) { + log.debug("[{}] skip sending topic policy event since the namespace is deleted", topicName); + return CompletableFuture.completedFuture(null); + } - SystemTopicClient systemTopicClient = - namespaceEventsSystemTopicFactory.createTopicPoliciesSystemTopicClient(topicName.getNamespaceObject()); + try { + createSystemTopicFactoryIfNeeded(); + } catch (PulsarServerException e) { + return CompletableFuture.failedFuture(e); + } - CompletableFuture> writerFuture = systemTopicClient.newWriterAsync(); - writerFuture.whenComplete((writer, ex) -> { - if (ex != null) { - result.completeExceptionally(ex); - } else { - PulsarEvent event = getPulsarEvent(topicName, actionType, policies); - CompletableFuture actionFuture = - ActionType.DELETE.equals(actionType) ? writer.deleteAsync(getEventKey(event), event) - : writer.writeAsync(getEventKey(event), event); - actionFuture.whenComplete(((messageId, e) -> { - if (e != null) { - result.completeExceptionally(e); - } else { - if (messageId != null) { - result.complete(null); - } else { - result.completeExceptionally(new RuntimeException("Got message id is null.")); - } - } - writer.closeAsync().whenComplete((v, cause) -> { - if (cause != null) { - log.error("[{}] Close writer error.", topicName, cause); + SystemTopicClient systemTopicClient = namespaceEventsSystemTopicFactory + .createTopicPoliciesSystemTopicClient(topicName.getNamespaceObject()); + + return systemTopicClient.newWriterAsync() + .thenCompose(writer -> { + PulsarEvent event = getPulsarEvent(topicName, actionType, policies); + CompletableFuture writeFuture = + ActionType.DELETE.equals(actionType) ? writer.deleteAsync(getEventKey(event), event) + : writer.writeAsync(getEventKey(event), event); + return writeFuture.handle((messageId, e) -> { + if (e != null) { + return CompletableFuture.failedFuture(e); } else { - if (log.isDebugEnabled()) { - log.debug("[{}] Close writer success.", topicName); + if (messageId != null) { + return CompletableFuture.completedFuture(null); + } else { + return CompletableFuture.failedFuture( + new RuntimeException("Got message id is null.")); } } - }); - }) - ); - } - }); - return result; + }).thenRun(() -> + writer.closeAsync().whenComplete((v, cause) -> { + if (cause != null) { + log.error("[{}] Close writer error.", topicName, cause); + } else { + if (log.isDebugEnabled()) { + log.debug("[{}] Close writer success.", topicName); + } + } + }) + ); + }); + }); } private PulsarEvent getPulsarEvent(TopicName topicName, ActionType actionType, TopicPolicies policies) { @@ -390,25 +393,25 @@ private void cleanCacheAndCloseReader(@Nonnull NamespaceName namespace, boolean private void readMorePolicies(SystemTopicClient.Reader reader) { reader.readNextAsync() - .thenAccept(msg -> { - refreshTopicPoliciesCache(msg); - notifyListener(msg); - }) - .whenComplete((__, ex) -> { - if (ex == null) { - readMorePolicies(reader); - } else { - Throwable cause = FutureUtil.unwrapCompletionException(ex); - if (cause instanceof PulsarClientException.AlreadyClosedException) { - log.warn("Read more topic policies exception, close the read now!", ex); - cleanCacheAndCloseReader( - reader.getSystemTopic().getTopicName().getNamespaceObject(), false); - } else { - log.warn("Read more topic polices exception, read again.", ex); - readMorePolicies(reader); - } - } - }); + .thenAccept(msg -> { + refreshTopicPoliciesCache(msg); + notifyListener(msg); + }) + .whenComplete((__, ex) -> { + if (ex == null) { + readMorePolicies(reader); + } else { + Throwable cause = FutureUtil.unwrapCompletionException(ex); + if (cause instanceof PulsarClientException.AlreadyClosedException) { + log.warn("Read more topic policies exception, close the read now!", ex); + cleanCacheAndCloseReader( + reader.getSystemTopic().getTopicName().getNamespaceObject(), false); + } else { + log.warn("Read more topic polices exception, read again.", ex); + readMorePolicies(reader); + } + } + }); } private void refreshTopicPoliciesCache(Message msg) { @@ -477,7 +480,7 @@ private boolean hasReplicateTo(Message message) { if (message instanceof MessageImpl) { return ((MessageImpl) message).hasReplicateTo() ? (((MessageImpl) message).getReplicateTo().size() == 1 - ? !((MessageImpl) message).getReplicateTo().contains(clusterName) : true) + ? !((MessageImpl) message).getReplicateTo().contains(clusterName) : true) : false; } if (message instanceof TopicMessageImpl) { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java index 7bb35cc4d63a8..64bbca0d3ed1d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java @@ -1648,6 +1648,11 @@ public void testDeleteNamespaceWithTopicPolicies() throws Exception { // create namespace2 String namespace = tenant + "/test-ns2"; + admin.namespaces().createNamespace(namespace, Set.of("test")); + admin.topics().createNonPartitionedTopic(namespace + "/tobedeleted"); + // verify namespace can be deleted even without topic policy events + admin.namespaces().deleteNamespace(namespace, true); + admin.namespaces().createNamespace(namespace, Set.of("test")); // create topic String topic = namespace + "/test-topic2"; @@ -1872,7 +1877,6 @@ public void testListOfNamespaceBundles() throws Exception { @Test public void testForceDeleteNamespace() throws Exception { - conf.setForceDeleteNamespaceAllowed(true); final String namespaceName = "prop-xyz2/ns1"; TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("test")); admin.tenants().createTenant("prop-xyz2", tenantInfo); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java index b303386c65ffd..fec0ce2da6ecf 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java @@ -167,6 +167,10 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest { @BeforeClass @Override public void setup() throws Exception { + setupConfigAndStart(null); + } + + private void applyDefaultConfig() { conf.setSystemTopicEnabled(false); conf.setTopicLevelPoliciesEnabled(false); conf.setLoadBalancerEnabled(true); @@ -178,6 +182,13 @@ public void setup() throws Exception { conf.setSubscriptionExpiryCheckIntervalInMinutes(1); conf.setBrokerDeleteInactiveTopicsEnabled(false); conf.setNumExecutorThreadPoolSize(5); + } + + private void setupConfigAndStart(java.util.function.Consumer configurationConsumer) throws Exception { + applyDefaultConfig(); + if (configurationConsumer != null) { + configurationConsumer.accept(conf); + } super.internalSetup(); @@ -215,6 +226,7 @@ public void resetClusters() throws Exception { pulsar.getConfiguration().setForceDeleteNamespaceAllowed(false); resetConfig(); + applyDefaultConfig(); setupClusters(); } @@ -1718,9 +1730,9 @@ public void testNamespaceSplitBundleWithInvalidAlgorithm() { @Test public void testNamespaceSplitBundleWithDefaultTopicCountEquallyDivideAlgorithm() throws Exception { cleanup(); - setup(); + setupConfigAndStart(conf -> conf + .setDefaultNamespaceBundleSplitAlgorithm(NamespaceBundleSplitAlgorithm.TOPIC_COUNT_EQUALLY_DIVIDE)); - conf.setDefaultNamespaceBundleSplitAlgorithm(NamespaceBundleSplitAlgorithm.TOPIC_COUNT_EQUALLY_DIVIDE); // Force to create a topic final String namespace = "prop-xyz/ns1"; List topicNames = Lists.newArrayList( @@ -1756,7 +1768,6 @@ public void testNamespaceSplitBundleWithDefaultTopicCountEquallyDivideAlgorithm( for (Producer producer : producers) { producer.close(); } - conf.setDefaultNamespaceBundleSplitAlgorithm(NamespaceBundleSplitAlgorithm.RANGE_EQUALLY_DIVIDE_NAME); } @Test @@ -1911,9 +1922,6 @@ public void testNamespaceUnloadBundle() throws Exception { @Test(dataProvider = "numBundles") public void testNamespaceBundleUnload(Integer numBundles) throws Exception { - cleanup(); - setup(); - admin.namespaces().createNamespace("prop-xyz/ns1-bundles", numBundles); admin.namespaces().setNamespaceReplicationClusters("prop-xyz/ns1-bundles", Set.of("test")); @@ -3259,7 +3267,8 @@ public void testBacklogSizeShouldBeZeroWhenConsumerAckedAllMessages() throws Exc @Test public void testGetTtlDurationDefaultInSeconds() throws Exception { - conf.setTtlDurationDefaultInSeconds(3600); + cleanup(); + setupConfigAndStart(conf -> conf.setTtlDurationDefaultInSeconds(3600)); Integer seconds = admin.namespaces().getPolicies("prop-xyz/ns1").message_ttl_in_seconds; assertNull(seconds); } @@ -3309,8 +3318,11 @@ public void testPartitionedTopicMsgDelayedAggregated() throws Exception { final String topic = "persistent://prop-xyz/ns1/testPartitionedTopicMsgDelayedAggregated-" + UUID.randomUUID().toString(); final String subName = "my-sub"; final int numPartitions = 2; - conf.setSubscriptionRedeliveryTrackerEnabled(true); - conf.setDelayedDeliveryEnabled(true); + cleanup(); + setupConfigAndStart(conf -> { + conf.setSubscriptionRedeliveryTrackerEnabled(true); + conf.setDelayedDeliveryEnabled(true); + }); admin.topics().createPartitionedTopic(topic, numPartitions); for (int i = 0; i < 2; i++) { @@ -3367,7 +3379,7 @@ public void testPartitionedTopicMsgDelayedAggregated() throws Exception { @Test(timeOut = 20000) public void testPartitionedTopicTruncate() throws Exception { - final String topicName = "persistent://prop-xyz/ns1/testTruncateTopic-" + UUID.randomUUID().toString(); + final String topicName = "persistent://prop-xyz/ns1/testTruncateTopic2-" + UUID.randomUUID().toString(); final String subName = "my-sub"; admin.topics().createPartitionedTopic(topicName,6); admin.namespaces().setRetention("prop-xyz/ns1", new RetentionPolicies(60, 50)); @@ -3387,9 +3399,13 @@ public void testPartitionedTopicTruncate() throws Exception { @Test(timeOut = 20000) public void testNonPartitionedTopicTruncate() throws Exception { - final String topicName = "persistent://prop-xyz/ns1/testTruncateTopic-" + UUID.randomUUID().toString(); + final String topicName = "persistent://prop-xyz/ns1/testTruncateTopic1-" + UUID.randomUUID().toString(); final String subName = "my-sub"; - this.conf.setTopicLevelPoliciesEnabled(true); + cleanup(); + setupConfigAndStart(conf -> { + conf.setTopicLevelPoliciesEnabled(true); + conf.setSystemTopicEnabled(true); + }); admin.topics().createNonPartitionedTopic(topicName); admin.namespaces().setRetention("prop-xyz/ns1", new RetentionPolicies(60, 50)); List messageIds = publishMessagesOnPersistentTopic(topicName, 10); @@ -3405,7 +3421,7 @@ public void testNonPartitionedTopicTruncate() throws Exception { @Test(timeOut = 20000) public void testNonPersistentTopicTruncate() throws Exception { - final String topicName = "non-persistent://prop-xyz/ns1/testTruncateTopic-" + UUID.randomUUID().toString(); + final String topicName = "non-persistent://prop-xyz/ns1/testTruncateTopic3-" + UUID.randomUUID().toString(); admin.topics().createNonPartitionedTopic(topicName); assertThrows(() -> {admin.topics().truncate(topicName);}); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java index 7b66b6a6b5152..f9fc717a817af 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java @@ -373,4 +373,15 @@ public void run() { } }); } + + @Test + public void testHandleNamespaceBeingDeleted() throws Exception { + SystemTopicBasedTopicPoliciesService service = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService(); + pulsar.getPulsarResources().getNamespaceResources().setPolicies(NamespaceName.get(NAMESPACE1), + old -> { + old.deleted = true; + return old; + }); + service.deleteTopicPoliciesAsync(TOPIC1).get(); + } }