[improve][broker] Use shrink map for message redelivery. (apache#15342)

(cherry picked from commit 615f05a) (cherry picked from commit b4762d1)
datastax · May 4, 2022 · 8e05428 · 8e05428
1 parent b300020
commit 8e05428
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 4 deletions.
diff --git a/...rc/main/java/org/apache/pulsar/broker/service/persistent/MessageRedeliveryController.java b/...rc/main/java/org/apache/pulsar/broker/service/persistent/MessageRedeliveryController.java
@@ -26,8 +26,8 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 import org.apache.bookkeeper.mledger.impl.PositionImpl;
-import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
-import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap.LongPair;
+import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap.LongPair;
 import org.apache.pulsar.common.util.collections.ConcurrentSortedLongPairSet;
 import org.apache.pulsar.common.util.collections.LongPairSet;
 
@@ -37,7 +37,10 @@ public class MessageRedeliveryController {
 
     public MessageRedeliveryController(boolean allowOutOfOrderDelivery) {
         this.messagesToRedeliver = new ConcurrentSortedLongPairSet(128, 2);
-        this.hashesToBeBlocked = allowOutOfOrderDelivery ? null : new ConcurrentLongLongPairHashMap(128, 2);
+        this.hashesToBeBlocked = allowOutOfOrderDelivery
+                ? null
+                : ConcurrentLongLongPairHashMap
+                    .newBuilder().concurrencyLevel(2).expectedItems(128).autoShrink(true).build();
     }
 
     public boolean add(long ledgerId, long entryId) {

diff --git a/...est/java/org/apache/pulsar/broker/service/persistent/MessageRedeliveryControllerTest.java b/...est/java/org/apache/pulsar/broker/service/persistent/MessageRedeliveryControllerTest.java
@@ -30,7 +30,7 @@
 import java.util.Set;
 import java.util.TreeSet;
 import org.apache.bookkeeper.mledger.impl.PositionImpl;
-import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap;
 import org.apache.pulsar.common.util.collections.LongPairSet;
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;