[fix][broker] Fix updatePartitionedTopic when replication at ns level…

… and topic policy is set (apache#22971) (cherry picked from commit 1c44fbb)
heesung-sn · Jun 25, 2024 · 153d7aa · 153d7aa
1 parent c9669bd
commit 153d7aa
Show file tree

Hide file tree

Showing 3 changed files with 32 additions and 9 deletions.
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
@@ -35,7 +35,6 @@
 import java.util.Base64;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -458,8 +457,8 @@ protected CompletableFuture<Void> internalCreateNonPartitionedTopicAsync(boolean
                                 Set<String> replicationClusters = policies.get().replication_clusters;
                                 TopicPolicies topicPolicies =
                                         pulsarService.getTopicPoliciesService().getTopicPoliciesIfExists(topicName);
-                                if (topicPolicies != null) {
-                                    replicationClusters = new HashSet<>(topicPolicies.getReplicationClusters());
+                                if (topicPolicies != null && topicPolicies.getReplicationClusters() != null) {
+                                    replicationClusters = topicPolicies.getReplicationClustersSet();
                                 }
                                 // Do check replicated clusters.
                                 if (replicationClusters.size() == 0) {

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
@@ -68,11 +68,11 @@
 import org.apache.pulsar.client.impl.ProducerImpl;
 import org.apache.pulsar.client.impl.PulsarClientImpl;
 import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
-import org.apache.pulsar.common.policies.data.TopicStats;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
+import org.apache.pulsar.common.policies.data.TopicStats;
 import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
 import org.awaitility.Awaitility;
 import org.awaitility.reflect.WhiteboxImpl;
 import org.mockito.Mockito;
@@ -456,6 +456,29 @@ public void testPartitionedTopicLevelReplication() throws Exception {
         admin2.topics().deletePartitionedTopic(topicName);
     }
 
+    // https://github.com/apache/pulsar/issues/22967
+    @Test
+    public void testPartitionedTopicWithTopicPolicyAndNoReplicationClusters() throws Exception {
+        final String topicName = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp_");
+        admin1.topics().createPartitionedTopic(topicName, 2);
+        try {
+            admin1.topicPolicies().setMessageTTL(topicName, 5);
+            Awaitility.await().ignoreExceptions().untilAsserted(() -> {
+                assertEquals(admin2.topics().getPartitionedTopicMetadata(topicName).partitions, 2);
+            });
+            admin1.topics().updatePartitionedTopic(topicName, 3, false);
+            Awaitility.await().ignoreExceptions().untilAsserted(() -> {
+                assertEquals(admin2.topics().getPartitionedTopicMetadata(topicName).partitions, 3);
+            });
+        } finally {
+            // cleanup.
+            admin1.topics().deletePartitionedTopic(topicName, true);
+            if (!usingGlobalZK) {
+                admin2.topics().deletePartitionedTopic(topicName, true);
+            }
+        }
+    }
+
     @Test
     public void testPartitionedTopicLevelReplicationRemoteTopicExist() throws Exception {
         final String topicName = BrokerTestUtil.newUniqueName("persistent://" + nonReplicatedNamespace + "/tp_");

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
@@ -266,6 +266,7 @@ protected void setConfigDefaults(ServiceConfiguration config, String clusterName
         config.setEnableReplicatedSubscriptions(true);
         config.setReplicatedSubscriptionsSnapshotFrequencyMillis(1000);
         config.setLoadBalancerSheddingEnabled(false);
+        config.setForceDeleteNamespaceAllowed(true);
     }
 
     @Override
@@ -276,11 +277,11 @@ protected void cleanup() throws Exception {
         if (!usingGlobalZK) {
             admin2.namespaces().setNamespaceReplicationClusters(replicatedNamespace, Sets.newHashSet(cluster2));
         }
-        admin1.namespaces().deleteNamespace(replicatedNamespace);
-        admin1.namespaces().deleteNamespace(nonReplicatedNamespace);
+        admin1.namespaces().deleteNamespace(replicatedNamespace, true);
+        admin1.namespaces().deleteNamespace(nonReplicatedNamespace, true);
         if (!usingGlobalZK) {
-            admin2.namespaces().deleteNamespace(replicatedNamespace);
-            admin2.namespaces().deleteNamespace(nonReplicatedNamespace);
+            admin2.namespaces().deleteNamespace(replicatedNamespace, true);
+            admin2.namespaces().deleteNamespace(nonReplicatedNamespace, true);
         }
 
         // shutdown.