[fix][broker] fix MessageDeduplication throw NPE when enable broker dedup and set namespace disable deduplication. (#20905)

Author: lifepuzzlefun

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java

@@ -482,6 +482,10 @@ public long getLastPublishedSequenceId(String producerName) {
     }
 
     public void takeSnapshot() {
+        if (!isEnabled()) {
+            return;
+        }
+
         Integer interval = topic.getHierarchyTopicPolicies().getDeduplicationSnapshotIntervalSeconds().get();
         long currentTimeStamp = System.currentTimeMillis();
         if (interval == null || interval <= 0

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/TopicDuplicationTest.java

@@ -492,6 +492,43 @@ public void testDisableNamespacePolicyTakeSnapshot() throws Exception {
 
     }
 
+    @Test(timeOut = 30000)
+    public void testDisableNamespacePolicyTakeSnapshotShouldNotThrowException() throws Exception {
+        cleanup();
+        conf.setBrokerDeduplicationEnabled(true);
+        conf.setBrokerDeduplicationSnapshotFrequencyInSeconds(1);
+        conf.setBrokerDeduplicationSnapshotIntervalSeconds(1);
+        conf.setBrokerDeduplicationEntriesInterval(20000);
+        setup();
+
+        final String topicName = testTopic + UUID.randomUUID().toString();
+        final String producerName = "my-producer";
+        @Cleanup
+        Producer<String> producer = pulsarClient
+                .newProducer(Schema.STRING).topic(topicName).enableBatching(false).producerName(producerName).create();
+
+        // disable deduplication
+        admin.namespaces().setDeduplicationStatus(myNamespace, false);
+
+        int msgNum = 50;
+        CountDownLatch countDownLatch = new CountDownLatch(msgNum);
+        for (int i = 0; i < msgNum; i++) {
+            producer.newMessage().value("msg" + i).sendAsync().whenComplete((res, e) -> countDownLatch.countDown());
+        }
+        countDownLatch.await();
+        PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService()
+                .getTopicIfExists(topicName).get().get();
+        ManagedCursor managedCursor = persistentTopic.getMessageDeduplication().getManagedCursor();
+
+        // when disable topic deduplication the cursor should be deleted.
+        assertNull(managedCursor);
+
+        // this method will be called at brokerService forEachTopic.
+        // if topic level disable deduplication.
+        // this method should be skipped without throw exception.
+        persistentTopic.checkDeduplicationSnapshot();
+    }
+
     private void waitCacheInit(String topicName) throws Exception {
         pulsarClient.newConsumer().topic(topicName).subscriptionName("my-sub").subscribe().close();
         TopicName topic = TopicName.get(topicName);