From 6ab493cf412a996d05f1af2cf50e1e1544bb2c88 Mon Sep 17 00:00:00 2001
From: feynmanlin <feynmanlin@tencent.com>
Date: Wed, 26 May 2021 15:16:58 +0800
Subject: [PATCH] Fix NPE when filtering read entries (#10704)

### Motivation
When ConcurrentOpenLongPairRangeSet does not contain any data, calling span() will throw NPE

### Modifications
return null instead
---
 .../apache/bookkeeper/mledger/impl/ManagedCursorImpl.java   | 4 ++--
 .../util/collections/ConcurrentOpenLongPairRangeSet.java    | 3 +++
 .../collections/ConcurrentOpenLongPairRangeSetTest.java     | 6 ++++++
 3 files changed, 11 insertions(+), 2 deletions(-)

diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index 51206c67a06fe..809e1e5c74f9b 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -2042,8 +2042,8 @@ List<Entry> filterReadEntries(List<Entry> entries) {
                 log.debug("[{}] [{}] Filtering entries {} - alreadyDeleted: {}", ledger.getName(), name, entriesRange,
                         individualDeletedMessages);
             }
-
-            if (individualDeletedMessages.isEmpty() || !entriesRange.isConnected(individualDeletedMessages.span())) {
+            if (individualDeletedMessages.isEmpty() || individualDeletedMessages.span() == null ||
+                    !entriesRange.isConnected(individualDeletedMessages.span())) {
                 // There are no individually deleted messages in this entry list, no need to perform filtering
                 if (log.isDebugEnabled()) {
                     log.debug("[{}] [{}] No filtering needed for entries {}", ledger.getName(), name, entriesRange);
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenLongPairRangeSet.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenLongPairRangeSet.java
index 34cb3280f9907..174f318aa1b61 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenLongPairRangeSet.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenLongPairRangeSet.java
@@ -171,6 +171,9 @@ public void clear() {
 
     @Override
     public Range<T> span() {
+        if (rangeBitSetMap.size() == 0) {
+            return null;
+        }
         Entry<Long, BitSet> firstSet = rangeBitSetMap.firstEntry();
         Entry<Long, BitSet> lastSet = rangeBitSetMap.lastEntry();
         int first = firstSet.getValue().nextSetBit(0);
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenLongPairRangeSetTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenLongPairRangeSetTest.java
index 04493fe7559e8..f57b75d52e10c 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenLongPairRangeSetTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenLongPairRangeSetTest.java
@@ -114,6 +114,12 @@ public void testAddCompareCompareWithGuava() {
         }
     }
 
+    @Test
+    public void testNPE() {
+        ConcurrentOpenLongPairRangeSet<LongPair> set = new ConcurrentOpenLongPairRangeSet<>(consumer);
+        assertNull(set.span());
+    }
+
     @Test
     public void testDeleteCompareWithGuava() {