diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java index 8d49d20553c74..ed15b964f1ede 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java @@ -429,6 +429,35 @@ private CompletableFuture delete(boolean failIfHasSubscriptions, boolean c return deleteFuture; } + private CompletableFuture tryToDeletePartitionedMetadata() { + if (TopicName.get(topic).isPartitioned() && !deletePartitionedTopicMetadataWhileInactive()) { + return CompletableFuture.completedFuture(null); + } + TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName()); + String path = AdminResource.path(AdminResource.PARTITIONED_TOPIC_PATH_ZNODE, topicName.getNamespace() + , topicName.getDomain().value(), topicName.getEncodedLocalName()); + try { + if (!getBrokerService().pulsar().getGlobalZkCache().exists(path)) { + return CompletableFuture.completedFuture(null); + } + CompletableFuture deleteMetadataFuture = new CompletableFuture<>(); + getBrokerService().pulsar().getGlobalZkCache().getZooKeeper().delete(path, -1 + , (rc, s, o) -> { + if (KeeperException.Code.OK.intValue() == rc + || KeeperException.Code.NONODE.intValue() == rc) { + getBrokerService().pulsar().getGlobalZkCache().invalidate(path); + deleteMetadataFuture.complete(null); + } else { + deleteMetadataFuture.completeExceptionally( + KeeperException.create(KeeperException.Code.get(rc))); + } + }, null); + return deleteMetadataFuture; + } catch (Exception e) { + return FutureUtil.failedFuture(e); + } + } + /** * Close this topic - close all producers and subscriptions associated with this topic * @@ -865,6 +894,7 @@ public void checkGC() { } stopReplProducers().thenCompose(v -> delete(true, false, true)) + .thenAccept(__ -> tryToDeletePartitionedMetadata()) .thenRun(() -> log.info("[{}] Topic deleted successfully due to inactivity", topic)) .exceptionally(e -> { if (e.getCause() instanceof TopicBusyException) { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/NonPersistentTopicE2ETest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/NonPersistentTopicE2ETest.java index fab7a5e1e03e5..bd2299953dba5 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/NonPersistentTopicE2ETest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/NonPersistentTopicE2ETest.java @@ -55,6 +55,8 @@ public class NonPersistentTopicE2ETest extends BrokerTestBase { @BeforeMethod @Override protected void setup() throws Exception { + conf.setBrokerDeleteInactivePartitionedTopicMetadataEnabled(true); + conf.setBrokerDeleteInactiveTopicsFrequencySeconds(1); super.baseSetup(); } @@ -231,5 +233,17 @@ public void testGC() throws Exception { producer2.close(); assertTrue(pulsar.getBrokerService().getTopicReference(topicName).isPresent()); + + // 6. Test for partitioned topic to delete the partitioned metadata + String topicGc = "non-persistent://prop/ns-abc/topic-gc"; + int partitions = 5; + admin.topics().createPartitionedTopic(topicGc, partitions); + Producer producer3 = pulsarClient.newProducer().topic(topicGc).create(); + producer3.close(); + assertTrue(pulsar.getBrokerService().fetchPartitionedTopicMetadataAsync( + TopicName.get(topicGc)).join().partitions == partitions); + runGC(); + assertTrue(pulsar.getBrokerService().fetchPartitionedTopicMetadataAsync( + TopicName.get(topicGc)).join().partitions == 0); } }