From 2007de6dedd6997cccd424f3d40988e99b0fa85a Mon Sep 17 00:00:00 2001
From: Fangbin Sun <sunfangbin@gmail.com>
Date: Fri, 20 Mar 2020 00:59:06 +0800
Subject: [PATCH] Fix NPE while call getLastMessageId. (#6562)

### Motivation

Fixes #6561

### Modifications

Initialize `BatchMessageAckerDisabled` with a `new BitSet()` Object.
---
 .../pulsar/client/impl/BatchMessageAcker.java |  2 +-
 .../impl/BatchMessageAckerDisabled.java       |  4 ++-
 .../client/impl/BatchMessageIdImplTest.java   | 30 +++++++++++++++++++
 3 files changed, 34 insertions(+), 2 deletions(-)

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageAcker.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageAcker.java
index 5df5ee15372ef..d46d3b38bf762 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageAcker.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageAcker.java
@@ -24,7 +24,7 @@
 class BatchMessageAcker {
 
     private BatchMessageAcker() {
-        this.bitSet = null;
+        this.bitSet = new BitSet();
         this.batchSize = 0;
     }
 
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageAckerDisabled.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageAckerDisabled.java
index a521d63446e82..538ee91ec1deb 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageAckerDisabled.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageAckerDisabled.java
@@ -18,12 +18,14 @@
  */
 package org.apache.pulsar.client.impl;
 
+import java.util.BitSet;
+
 class BatchMessageAckerDisabled extends BatchMessageAcker {
 
     static final BatchMessageAckerDisabled INSTANCE = new BatchMessageAckerDisabled();
 
     private BatchMessageAckerDisabled() {
-        super(null, 0);
+        super(new BitSet(), 0);
     }
 
     @Override
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BatchMessageIdImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BatchMessageIdImplTest.java
index 9e76a66b4d39a..c32a59376f9df 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BatchMessageIdImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BatchMessageIdImplTest.java
@@ -21,7 +21,11 @@
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNotEquals;
 import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectWriter;
+import org.apache.pulsar.common.util.ObjectMapperFactory;
 import org.testng.annotations.Test;
 
 public class BatchMessageIdImplTest {
@@ -68,4 +72,30 @@ public void equalsTest() {
         assertEquals(msgId, batchMsgId4);
     }
 
+    @Test
+    public void deserializationTest() {
+        // initialize BitSet with null
+        BatchMessageAcker ackerDisabled = new BatchMessageAcker(null, 0);
+        BatchMessageIdImpl batchMsgId = new BatchMessageIdImpl(0, 0, 0, 0, ackerDisabled);
+
+        ObjectWriter writer = ObjectMapperFactory.create().writerWithDefaultPrettyPrinter();
+
+        try {
+            writer.writeValueAsString(batchMsgId);
+            fail("Shouldn't be deserialized");
+        } catch (JsonProcessingException e) {
+            // expected
+            assertTrue(e.getCause() instanceof NullPointerException);
+        }
+
+        // use the default BatchMessageAckerDisabled
+        BatchMessageIdImpl batchMsgIdToDeserialize = new BatchMessageIdImpl(0, 0, 0, 0);
+
+        try {
+            writer.writeValueAsString(batchMsgIdToDeserialize);
+        } catch (JsonProcessingException e) {
+            fail("Should be successful");
+        }
+    }
+
 }