From 4b2c6738ef15a29e90a7f5362cbfb8769f8ac80f Mon Sep 17 00:00:00 2001
From: feynmanlin <feynmanlin@tencent.com>
Date: Thu, 27 May 2021 01:37:21 +0800
Subject: [PATCH] Fix inconsistent behavior in LongPairRangeSet (#10713)

* fix inconsistent behavior

* add unit test

* fix unit test

* add unit test
---
 .../util/collections/LongPairRangeSet.java    | 14 ++++-
 .../util/collections/DefaultRangeSetTest.java | 63 +++++++++++++++++++
 2 files changed, 75 insertions(+), 2 deletions(-)
 create mode 100644 pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/DefaultRangeSetTest.java

diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/LongPairRangeSet.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/LongPairRangeSet.java
index 92ae3a659201b..87ffda299d479 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/LongPairRangeSet.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/LongPairRangeSet.java
@@ -26,7 +26,9 @@
 import lombok.EqualsAndHashCode;
 
 import java.util.Collection;
+import java.util.Iterator;
 import java.util.List;
+import java.util.NoSuchElementException;
 import java.util.Set;
 
 /**
@@ -239,7 +241,11 @@ public boolean isEmpty() {
 
         @Override
         public Range<T> span() {
-            return set.span();
+            try {
+                return set.span();
+            } catch (NoSuchElementException e) {
+                return null;
+            }
         }
 
         @Override
@@ -268,7 +274,11 @@ public boolean contains(long key, long value) {
 
         @Override
         public Range<T> firstRange() {
-            return set.asRanges().iterator().next();
+            Iterator<Range<T>> iterable = set.asRanges().iterator();
+            if (iterable.hasNext()) {
+                return set.asRanges().iterator().next();
+            }
+            return null;
         }
 
         @Override
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/DefaultRangeSetTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/DefaultRangeSetTest.java
new file mode 100644
index 0000000000000..e2fc5c1d32d72
--- /dev/null
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/DefaultRangeSetTest.java
@@ -0,0 +1,63 @@
+/**
+ * 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.common.util.collections;
+
+import org.testng.annotations.Test;
+
+import static org.testng.AssertJUnit.assertNull;
+import static org.testng.AssertJUnit.assertTrue;
+import static org.testng.AssertJUnit.fail;
+
+public class DefaultRangeSetTest {
+    static final LongPairRangeSet.LongPairConsumer<LongPairRangeSet.LongPair> consumer =
+            LongPairRangeSet.LongPair::new;
+
+    @Test
+    public void testBehavior() {
+        LongPairRangeSet.DefaultRangeSet<LongPairRangeSet.LongPair> set =
+                new LongPairRangeSet.DefaultRangeSet<>(consumer);
+        ConcurrentOpenLongPairRangeSet<LongPairRangeSet.LongPair> rangeSet =
+                new ConcurrentOpenLongPairRangeSet<>(consumer);
+
+        assertNull(set.firstRange());
+        assertNull(set.lastRange());
+        assertNull(set.span());
+
+        assertNull(rangeSet.firstRange());
+        assertNull(rangeSet.lastRange());
+        assertNull(rangeSet.span());
+
+        try {
+            set.contains(null);
+            fail("should fail");
+        } catch (NullPointerException ignore) {
+        }
+        try {
+            rangeSet.contains(null);
+            fail("should fail");
+        } catch (NullPointerException ignore) {
+        }
+
+        rangeSet.addOpenClosed(9, 0, 10, 10);
+        set.addOpenClosed(9, 0, 10, 10);
+        assertTrue(rangeSet.contains(10,1));
+        assertTrue(set.contains(10,1));
+
+    }
+}