This is an automated email from the ASF dual-hosted git repository. lhotari pushed a commit to branch branch-3.0 in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit df02f0b97a5bff37149c2ae358fc099dfd8d009d Author: fengyubiao <[email protected]> AuthorDate: Wed Apr 9 14:29:31 2025 +0800 [fix][broker] The feature brokerDeleteInactivePartitionedTopicMetadataEnabled leaves orphan topic policies and topic schemas (#24150) (cherry picked from commit 610b72e2fb71b3f48c542f906fbff1b457a74f24) --- conf/broker.conf | 5 +- .../apache/pulsar/broker/ServiceConfiguration.java | 8 +- .../broker/service/persistent/PersistentTopic.java | 10 +- .../apache/pulsar/broker/admin/AdminApi2Test.java | 14 ++ .../broker/service/OneWayReplicatorTestBase.java | 17 ++ .../broker/service/ReplicationTopicGcTest.java | 192 +++++++++++++++++++++ .../ReplicationTopicGcUsingGlobalZKTest.java | 63 +++++++ 7 files changed, 304 insertions(+), 5 deletions(-) diff --git a/conf/broker.conf b/conf/broker.conf index 627e968848d..a7868ce1e0f 100644 --- a/conf/broker.conf +++ b/conf/broker.conf @@ -213,8 +213,11 @@ brokerDeleteInactiveTopicsFrequencySeconds=60 brokerDeleteInactiveTopicsMode=delete_when_no_subscriptions # Metadata of inactive partitioned topic will not be cleaned up automatically by default. -# Note: If `allowAutoTopicCreation` and this option are enabled at the same time, +# Note 1: If `allowAutoTopicCreation` and this option are enabled at the same time, # it may appear that a partitioned topic has just been deleted but is automatically created as a non-partitioned topic. +# Note 2: Activating bidirectional geo-replication under global configuration ZooKeeper may lead to schema remnants and +# abnormal topic-level policies. +# Note 3: Activating bidirectional geo-replication under global configuration ZooKeeper may lead to a consumption issue. brokerDeleteInactivePartitionedTopicMetadataEnabled=false # Max duration of topic inactivity in seconds, default is not present diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java index d58679876ae..8979b949a84 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java @@ -670,9 +670,13 @@ public class ServiceConfiguration implements PulsarConfiguration { category = CATEGORY_POLICIES, dynamic = true, doc = "Metadata of inactive partitioned topic will not be automatically cleaned up by default.\n" - + "Note: If `allowAutoTopicCreation` and this option are enabled at the same time,\n" + + "Note 1: If `allowAutoTopicCreation` and this option are enabled at the same time,\n" + "it may appear that a partitioned topic has just been deleted but is automatically created as a " - + "non-partitioned topic." + + "non-partitioned topic.\n" + + "Note 2: Activating bidirectional geo-replication under global ZooKeeper configuration may lead to schema" + + " remnants and abnormal topic-level policies.\n" + + "Note 3: Activating bidirectional geo-replication under global configuration ZooKeeper may lead" + + " to a consumption issue." ) private boolean brokerDeleteInactivePartitionedTopicMetadataEnabled = false; @FieldContext( 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 acf51d645e2..4e71237e30c 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 @@ -2986,8 +2986,14 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal String.format("Another partition exists for [%s].", topicName)); } else { - return partitionedTopicResources - .deletePartitionedTopicAsync(topicName); + try { + return brokerService.getPulsar().getAdminClient().topics() + .deletePartitionedTopicAsync(topicName.toString()); + } catch (PulsarServerException e) { + log.info("[{}] Delete topic metadata failed due to failed to" + + " get internal admin client.", topicName, e); + return CompletableFuture.failedFuture(e); + } } }); })) 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 385c2677899..c8c5a683153 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 @@ -3660,4 +3660,18 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest { List.of(".*", "broker.*") ); } + + @Test + public void testDeletePatchyPartitionedTopic() throws Exception { + final String topic = BrokerTestUtil.newUniqueName(defaultNamespace + "/tp"); + admin.topics().createPartitionedTopic(topic, 2); + Producer producer = pulsarClient.newProducer().topic(TopicName.get(topic).getPartition(0).toString()) + .create(); + // Mock a scenario that "-partition-1" has been removed due to topic GC. + pulsar.getBrokerService().getTopic(TopicName.get(topic).getPartition(1).toString(), false) + .get().get().delete().join(); + // Verify: delete partitioned topic. + producer.close(); + admin.topics().deletePartitionedTopicAsync(topic, false).get(); + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java index f3076ebdec6..508808eac3b 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java @@ -480,4 +480,21 @@ public abstract class OneWayReplicatorTestBase extends TestRetrySupport { admin2.topics().delete(topicName.toString()); } } + + protected void waitReplicatorStopped(PulsarService sourceCluster, PulsarService targetCluster, String topicName) { + Awaitility.await().atMost(Duration.ofSeconds(60)).untilAsserted(() -> { + Optional<Topic> topicOptional2 = targetCluster.getBrokerService().getTopic(topicName, false).get(); + assertTrue(topicOptional2.isPresent()); + PersistentTopic persistentTopic2 = (PersistentTopic) topicOptional2.get(); + for (org.apache.pulsar.broker.service.Producer producer : persistentTopic2.getProducers().values()) { + assertFalse(producer.getProducerName() + .startsWith(targetCluster.getConfiguration().getReplicatorPrefix())); + } + Optional<Topic> topicOptional1 = sourceCluster.getBrokerService().getTopic(topicName, false).get(); + assertTrue(topicOptional1.isPresent()); + PersistentTopic persistentTopic1 = (PersistentTopic) topicOptional1.get(); + assertTrue(persistentTopic1.getReplicators().isEmpty() + || !persistentTopic1.getReplicators().get(targetCluster.getConfig().getClusterName()).isConnected()); + }); + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTopicGcTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTopicGcTest.java new file mode 100644 index 00000000000..01ace3672f6 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTopicGcTest.java @@ -0,0 +1,192 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.service; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertTrue; +import java.util.Collections; +import java.util.concurrent.TimeUnit; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.collections4.CollectionUtils; +import org.apache.pulsar.broker.BrokerTestUtil; +import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.common.policies.data.InactiveTopicDeleteMode; +import org.apache.pulsar.common.policies.data.PublishRate; +import org.apache.pulsar.common.policies.data.TopicType; +import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble; +import org.apache.pulsar.zookeeper.ZookeeperServerTest; +import org.awaitility.Awaitility; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +@Slf4j +@Test(groups = "broker") +public class ReplicationTopicGcTest extends OneWayReplicatorTestBase { + + @Override + @BeforeClass(alwaysRun = true, timeOut = 300000) + public void setup() throws Exception { + super.setup(); + } + + @Override + @AfterClass(alwaysRun = true, timeOut = 300000) + public void cleanup() throws Exception { + super.cleanup(); + } + + @DataProvider(name = "topicTypes") + public Object[][] topicTypes() { + return new Object[][]{ + {TopicType.NON_PARTITIONED}, + {TopicType.PARTITIONED} + }; + } + + protected void setConfigDefaults(ServiceConfiguration config, String clusterName, + LocalBookkeeperEnsemble bookkeeperEnsemble, ZookeeperServerTest brokerConfigZk) { + super.setConfigDefaults(config, clusterName, bookkeeperEnsemble, brokerConfigZk); + config.setBrokerDeleteInactiveTopicsMode(InactiveTopicDeleteMode.delete_when_no_subscriptions); + config.setBrokerDeleteInactiveTopicsEnabled(true); + config.setBrokerDeleteInactivePartitionedTopicMetadataEnabled(true); + config.setBrokerDeleteInactiveTopicsFrequencySeconds(5); + config.setBrokerDeleteInactiveTopicsMaxInactiveDurationSeconds(5); + config.setReplicationPolicyCheckDurationSeconds(1); + } + + @Test(dataProvider = "topicTypes") + public void testTopicGC(TopicType topicType) throws Exception { + final String topicName = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp_"); + final String schemaId = TopicName.get(topicName).getSchemaName(); + final String subTopic = topicType == TopicType.NON_PARTITIONED ? topicName + : TopicName.get(topicName).getPartition(0).toString(); + if (topicType == TopicType.NON_PARTITIONED) { + admin1.topics().createNonPartitionedTopic(topicName); + } else { + admin1.topics().createPartitionedTopic(topicName, 1); + } + + Producer<String> producer1 = client1.newProducer(Schema.STRING).topic(topicName).create(); + // Wait for replicator started. + waitReplicatorStarted(subTopic); + + // Trigger a topic level policies. + PublishRate publishRate = new PublishRate(1000, 1024 * 1024); + admin1.topicPolicies().setPublishRate(topicName, publishRate); + admin2.topicPolicies().setPublishRate(topicName, publishRate); + // Write a schema. + // Since there is a producer registered one the source cluster, skipped to write a schema. + admin2.schemas().createSchema(topicName, Schema.STRING.getSchemaInfo()); + + // Trigger GC through close all clients. + producer1.close(); + // Verify: all resources were deleted. + Awaitility.await().atMost(60, TimeUnit.SECONDS).untilAsserted(() -> { + // sub topic. + assertFalse(pulsar1.getPulsarResources().getTopicResources() + .persistentTopicExists(TopicName.get(subTopic)).get()); + assertFalse(pulsar2.getPulsarResources().getTopicResources() + .persistentTopicExists(TopicName.get(subTopic)).get()); + // partitioned topic. + assertFalse(pulsar1.getPulsarResources().getNamespaceResources() + .getPartitionedTopicResources().partitionedTopicExists(TopicName.get(topicName))); + assertFalse(pulsar2.getPulsarResources().getNamespaceResources() + .getPartitionedTopicResources().partitionedTopicExists(TopicName.get(topicName))); + // topic policies. + assertTrue(pulsar1.getTopicPoliciesService().getTopicPoliciesAsync(TopicName.get(topicName)) + .get().isEmpty()); + assertTrue(pulsar2.getTopicPoliciesService().getTopicPoliciesAsync(TopicName.get(topicName)) + .get().isEmpty()); + // schema. + assertTrue(CollectionUtils.isEmpty(pulsar1.getSchemaStorage().getAll(schemaId).get())); + assertTrue(CollectionUtils.isEmpty(pulsar2.getSchemaStorage().getAll(schemaId).get())); + }); + } + + @Test(dataProvider = "topicTypes") + public void testRemoteClusterStillConsumeAfterCurrentClusterGc(TopicType topicType) throws Exception { + final String topicName = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp_"); + final String subscription = "s1"; + final String schemaId = TopicName.get(topicName).getSchemaName(); + final String subTopic = topicType == TopicType.NON_PARTITIONED ? topicName + : TopicName.get(topicName).getPartition(0).toString(); + if (topicType == TopicType.NON_PARTITIONED) { + admin1.topics().createNonPartitionedTopic(topicName); + } else { + admin1.topics().createPartitionedTopic(topicName, 1); + } + + // Wait for replicator started. + Producer<String> producer1 = client1.newProducer(Schema.STRING).topic(topicName).create(); + waitReplicatorStarted(subTopic); + admin2.topics().createSubscription(topicName, subscription, MessageId.earliest); + + if (usingGlobalZK) { + admin2.topics().setReplicationClusters(topicName, Collections.singletonList(cluster2)); + waitReplicatorStopped(pulsar2, pulsar1, subTopic); + } + + // Send a message + producer1.send("msg-1"); + Awaitility.await().untilAsserted(() -> { + assertEquals(admin2.topics().getStats(subTopic).getSubscriptions().get(subscription).getMsgBacklog(), 1); + }); + + // Trigger GC through close all clients. + producer1.close(); + // Verify: the topic was removed on the source cluster. + Awaitility.await().atMost(60, TimeUnit.SECONDS).untilAsserted(() -> { + // sub topic. + assertFalse(pulsar1.getPulsarResources().getTopicResources() + .persistentTopicExists(TopicName.get(subTopic)).get()); + // partitioned topic. + assertFalse(pulsar1.getPulsarResources().getNamespaceResources() + .getPartitionedTopicResources().partitionedTopicExists(TopicName.get(topicName))); + // topic policies. + assertTrue(pulsar1.getTopicPoliciesService().getTopicPoliciesAsync(TopicName.get(topicName)) + .get().isEmpty()); + // schema. + assertTrue(CollectionUtils.isEmpty(pulsar1.getSchemaStorage().getAll(schemaId).get())); + }); + + // Verify: consumer still can consume messages from the remote cluster. + org.apache.pulsar.client.api.Consumer<String> consumer = + client2.newConsumer(Schema.STRING).topic(topicName).subscriptionName(subscription).subscribe(); + Message<String> msg = consumer.receive(2, TimeUnit.SECONDS); + assertNotNull(msg); + assertEquals(msg.getValue(), "msg-1"); + + // Cleanup. + consumer.close(); + if (topicType == TopicType.NON_PARTITIONED) { + admin2.topics().delete(topicName); + } else { + admin2.topics().deletePartitionedTopic(topicName); + } + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTopicGcUsingGlobalZKTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTopicGcUsingGlobalZKTest.java new file mode 100644 index 00000000000..ce6eddc5698 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTopicGcUsingGlobalZKTest.java @@ -0,0 +1,63 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.service; + +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.common.policies.data.TopicType; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +@Slf4j +@Test(groups = "broker") +public class ReplicationTopicGcUsingGlobalZKTest extends ReplicationTopicGcTest { + + @Override + @BeforeClass(alwaysRun = true, timeOut = 300000) + public void setup() throws Exception { + super.usingGlobalZK = true; + super.setup(); + } + + @Override + @AfterClass(alwaysRun = true, timeOut = 300000) + public void cleanup() throws Exception { + super.cleanup(); + } + + @Test(dataProvider = "topicTypes") + public void testTopicGC(TopicType topicType) throws Exception { + if (topicType.equals(TopicType.PARTITIONED)) { + // Pulsar does not support the feature "brokerDeleteInactivePartitionedTopicMetadataEnabled" when enabling + // Geo-Replication with Global ZK. + return; + } + super.testTopicGC(topicType); + } + + @Test(dataProvider = "topicTypes") + public void testRemoteClusterStillConsumeAfterCurrentClusterGc(TopicType topicType) throws Exception { + if (topicType.equals(TopicType.PARTITIONED)) { + // Pulsar does not support the feature "brokerDeleteInactivePartitionedTopicMetadataEnabled" when enabling + // Geo-Replication with Global ZK. + return; + } + super.testRemoteClusterStillConsumeAfterCurrentClusterGc(topicType); + } +}
