[pulsar] branch branch-2.9 updated: Fixes NPE - ``ReplicatedSubscriptionsController`` send marker message when enable deduplicated. (#14017)

[mm...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

mmerli pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-2.9 by this push:
     new affd047  Fixes NPE - ``ReplicatedSubscriptionsController`` send marker message when enable deduplicated. (#14017)
affd047 is described below

commit affd0476866abd6a50ae41bc4ee8a661f829d68d
Author: Qiang Zhao <74...@users.noreply.github.com>
AuthorDate: Sat Jan 29 01:43:53 2022 +0800

    Fixes NPE - ``ReplicatedSubscriptionsController`` send marker message when enable deduplicated. (#14017)
    
    * Fixes marker message cause NPE.
    
    * Fixes marker message cause NPE.
---
 .../service/persistent/MessageDeduplication.java   |  7 ++-
 .../broker/BrokerMessageDeduplicationTest.java     | 73 ++++++++++++++++++++++
 2 files changed, 77 insertions(+), 3 deletions(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java
index e2436bb..9913ddf 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java
@@ -79,7 +79,8 @@ public class MessageDeduplication {
         Failed,
     }
 
-    enum MessageDupStatus {
+    @VisibleForTesting
+    public enum MessageDupStatus {
         // whether a message is a definitely a duplicate or not cannot be determined at this time
         Unknown,
         // message is definitely NOT a duplicate
@@ -312,7 +313,7 @@ public class MessageDeduplication {
      * @return true if the message should be published or false if it was recognized as a duplicate
      */
     public MessageDupStatus isDuplicate(PublishContext publishContext, ByteBuf headersAndPayload) {
-        if (!isEnabled()) {
+        if (!isEnabled() || publishContext.isMarkerMessage()) {
             return MessageDupStatus.NotDup;
         }
 
@@ -365,7 +366,7 @@ public class MessageDeduplication {
      * Call this method whenever a message is persisted to get the chance to trigger a snapshot.
      */
     public void recordMessagePersisted(PublishContext publishContext, PositionImpl position) {
-        if (!isEnabled()) {
+        if (!isEnabled() || publishContext.isMarkerMessage()) {
             return;
         }
 
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerMessageDeduplicationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerMessageDeduplicationTest.java
new file mode 100644
index 0000000..f2d7a21
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerMessageDeduplicationTest.java
@@ -0,0 +1,73 @@
+/**
+ * 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;
+
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+import org.apache.bookkeeper.mledger.ManagedLedger;
+import org.apache.pulsar.broker.service.Topic;
+import org.apache.pulsar.broker.service.persistent.MessageDeduplication;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.testng.annotations.Test;
+
+public class BrokerMessageDeduplicationTest {
+
+    @Test
+    public void markerMessageNotDeduplicated() {
+        PulsarService pulsarService = mock(PulsarService.class);
+        ServiceConfiguration configuration = new ServiceConfiguration();
+        doReturn(configuration).when(pulsarService).getConfiguration();
+        MessageDeduplication deduplication = spy(new MessageDeduplication(pulsarService,
+                mock(PersistentTopic.class), mock(ManagedLedger.class)));
+        doReturn(true).when(deduplication).isEnabled();
+        Topic.PublishContext context = mock(Topic.PublishContext.class);
+        doReturn(true).when(context).isMarkerMessage();
+        MessageDeduplication.MessageDupStatus status = deduplication.isDuplicate(context, null);
+        assertEquals(status, MessageDeduplication.MessageDupStatus.NotDup);
+    }
+
+    @Test
+    public void markerMessageNotRecordPersistent() {
+        PulsarService pulsarService = mock(PulsarService.class);
+        ServiceConfiguration configuration = new ServiceConfiguration();
+        doReturn(configuration).when(pulsarService).getConfiguration();
+        MessageDeduplication deduplication = spy(new MessageDeduplication(pulsarService,
+                mock(PersistentTopic.class), mock(ManagedLedger.class)));
+        doReturn(true).when(deduplication).isEnabled();
+        Topic.PublishContext context = mock(Topic.PublishContext.class);
+         // marker message don't record message persisted.
+        doReturn(true).when(context).isMarkerMessage();
+        deduplication.recordMessagePersisted(context, null);
+
+        // if is not a marker message, we will get NPE. because context is mocked with null value fields.
+        doReturn(false).when(context).isMarkerMessage();
+        try {
+            deduplication.recordMessagePersisted(context, null);
+            fail();
+        } catch (Exception npe) {
+            assertTrue(npe instanceof NullPointerException);
+        }
+    }
+
+
+}