You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by ba...@apache.org on 2023/10/21 08:51:39 UTC
[pulsar] branch branch-2.11 updated (830acea4f44 -> edbb6bba03a)
This is an automated email from the ASF dual-hosted git repository.
baodi pushed a change to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 830acea4f44 [fix][broker] Fixed reset for AggregatedNamespaceStats (#21225)
new c2c9db3d626 [fix][broker] fix bug caused by optimistic locking (#18390)
new edbb6bba03a [fix][broker] fix ModularLoadManagerImpl always delete active bundle-data. sec ver. (#20620)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../loadbalance/impl/ModularLoadManagerImpl.java | 21 ++--
.../impl/ModularLoadManagerImplTest.java | 119 +++++++++++++++++++++
.../util/collections/ConcurrentLongHashMap.java | 25 +++--
.../collections/ConcurrentLongLongPairHashMap.java | 39 ++++---
.../util/collections/ConcurrentLongPairSet.java | 41 ++++---
.../util/collections/ConcurrentOpenHashMap.java | 39 ++++---
.../util/collections/ConcurrentOpenHashSet.java | 22 ++--
.../apache/pulsar/common/util/FutureUtilTest.java | 1 -
.../collections/ConcurrentLongHashMapTest.java | 62 ++++++++++-
.../ConcurrentLongLongPairHashMapTest.java | 65 +++++++++++
.../collections/ConcurrentLongPairSetTest.java | 64 +++++++++++
.../collections/ConcurrentOpenHashMapTest.java | 64 +++++++++++
.../collections/ConcurrentOpenHashSetTest.java | 64 +++++++++++
13 files changed, 543 insertions(+), 83 deletions(-)
[pulsar] 02/02: [fix][broker] fix ModularLoadManagerImpl always delete active bundle-data. sec ver. (#20620)
Posted by ba...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
baodi pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit edbb6bba03a4e3bb08bdb308ddd09a856028e963
Author: lifepuzzlefun <wj...@163.com>
AuthorDate: Mon Aug 21 10:39:28 2023 +0800
[fix][broker] fix ModularLoadManagerImpl always delete active bundle-data. sec ver. (#20620)
Co-authored-by: wangjinlong <wa...@zhihu.com>
(cherry picked from commit f2f0bf47c02ba8c15c63ec5cfb177f913b3fec9a)
---
.../loadbalance/impl/ModularLoadManagerImpl.java | 21 ++--
.../impl/ModularLoadManagerImplTest.java | 119 +++++++++++++++++++++
2 files changed, 129 insertions(+), 11 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
index daa91f7cc4a..d9d1629cc10 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
@@ -566,17 +566,6 @@ public class ModularLoadManagerImpl implements ModularLoadManager {
bundleData.put(bundle, currentBundleData);
}
}
-
- //Remove not active bundle from loadData
- for (String bundle : bundleData.keySet()) {
- if (!activeBundles.contains(bundle)){
- bundleData.remove(bundle);
- if (pulsar.getLeaderElectionService().isLeader()){
- deleteBundleDataFromMetadataStore(bundle);
- }
- }
- }
-
// Remove all loaded bundles from the preallocated maps.
final Map<String, BundleData> preallocatedBundleData = brokerData.getPreallocatedBundleData();
Set<String> ownedNsBundles = pulsar.getNamespaceService().getOwnedServiceUnits()
@@ -611,6 +600,16 @@ public class ModularLoadManagerImpl implements ModularLoadManager {
LoadManagerShared.fillNamespaceToBundlesMap(preallocatedBundleData.keySet(), namespaceToBundleRange);
}
}
+
+ // Remove not active bundle from loadData
+ for (String bundle : bundleData.keySet()) {
+ if (!activeBundles.contains(bundle)){
+ bundleData.remove(bundle);
+ if (pulsar.getLeaderElectionService().isLeader()){
+ deleteBundleDataFromMetadataStore(bundle);
+ }
+ }
+ }
}
/**
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java
index 013f27d0127..1779f1d1fe1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java
@@ -42,7 +42,9 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
@@ -60,17 +62,22 @@ import org.apache.pulsar.broker.loadbalance.LoadManager;
import org.apache.pulsar.broker.loadbalance.impl.LoadManagerShared.BrokerTopicLoadingPredicate;
import org.apache.pulsar.client.admin.Namespaces;
import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceBundleFactory;
+import org.apache.pulsar.common.naming.NamespaceBundleSplitAlgorithm;
import org.apache.pulsar.common.naming.NamespaceBundles;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.ServiceUnitId;
+import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.NamespaceIsolationDataImpl;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.util.ObjectMapperFactory;
+import org.apache.pulsar.metadata.api.MetadataCache;
import org.apache.pulsar.metadata.api.Notification;
import org.apache.pulsar.metadata.api.NotificationType;
import org.apache.pulsar.metadata.api.extended.CreateOption;
@@ -719,4 +726,116 @@ public class ModularLoadManagerImplTest {
}
+
+
+ @Test
+ public void testRemoveNonExistBundleData()
+ throws PulsarAdminException, InterruptedException,
+ PulsarClientException, PulsarServerException, NoSuchFieldException, IllegalAccessException {
+ final String cluster = "use";
+ final String tenant = "my-tenant";
+ final String namespace = "remove-non-exist-bundle-data-ns";
+ final String topicName = tenant + "/" + namespace + "/" + "topic";
+ int bundleNumbers = 8;
+
+ admin1.clusters().createCluster(cluster, ClusterData.builder().serviceUrl("http://" + pulsar1.getAdvertisedAddress()).build());
+ admin1.tenants().createTenant(tenant,
+ new TenantInfoImpl(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet(cluster)));
+ admin1.namespaces().createNamespace(tenant + "/" + namespace, bundleNumbers);
+
+ @Cleanup
+ PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(pulsar1.getBrokerServiceUrl()).build();
+
+ // create a lot of topic to fully distributed among bundles.
+ for (int i = 0; i < 10; i++) {
+ String topicNameI = topicName + i;
+ admin1.topics().createPartitionedTopic(topicNameI, 20);
+ // trigger bundle assignment
+
+ pulsarClient.newConsumer().topic(topicNameI)
+ .subscriptionName("my-subscriber-name2").subscribe();
+ }
+
+ ModularLoadManagerWrapper loadManagerWrapper = (ModularLoadManagerWrapper) pulsar1.getLoadManager().get();
+ ModularLoadManagerImpl lm1 = (ModularLoadManagerImpl) loadManagerWrapper.getLoadManager();
+ ModularLoadManagerWrapper loadManager2 = (ModularLoadManagerWrapper) pulsar2.getLoadManager().get();
+ ModularLoadManagerImpl lm2 = (ModularLoadManagerImpl) loadManager2.getLoadManager();
+
+ Field executors = lm1.getClass().getDeclaredField("executors");
+ executors.setAccessible(true);
+ ExecutorService executorService = (ExecutorService) executors.get(lm1);
+
+ assertEquals(lm1.getAvailableBrokers().size(), 2);
+
+ pulsar1.getBrokerService().updateRates();
+ pulsar2.getBrokerService().updateRates();
+
+ lm1.writeBrokerDataOnZooKeeper(true);
+ lm2.writeBrokerDataOnZooKeeper(true);
+
+ // wait for metadata store notification finish
+ CountDownLatch latch = new CountDownLatch(1);
+ executorService.submit(latch::countDown);
+ latch.await();
+
+ loadManagerWrapper.writeResourceQuotasToZooKeeper();
+
+ MetadataCache<BundleData> bundlesCache = pulsar1.getLocalMetadataStore().getMetadataCache(BundleData.class);
+
+ // trigger bundle split
+ String topicToFindBundle = topicName + 0;
+ NamespaceBundle bundleWillBeSplit = pulsar1.getNamespaceService().getBundle(TopicName.get(topicToFindBundle));
+
+ String bundleDataPath = ModularLoadManagerImpl.BUNDLE_DATA_PATH + "/" + tenant + "/" + namespace;
+ CompletableFuture<List<String>> children = bundlesCache.getChildren(bundleDataPath);
+ List<String> bundles = children.join();
+ assertTrue(bundles.contains(bundleWillBeSplit.getBundleRange()));
+
+ // after updateAll no namespace bundle data is deleted from metadata store.
+ lm1.updateAll();
+
+ children = bundlesCache.getChildren(bundleDataPath);
+ bundles = children.join();
+ assertFalse(bundles.isEmpty());
+ assertEquals(bundleNumbers, bundles.size());
+
+ NamespaceName namespaceName = NamespaceName.get(tenant, namespace);
+ pulsar1.getAdminClient().namespaces().splitNamespaceBundle(tenant + "/" + namespace,
+ bundleWillBeSplit.getBundleRange(),
+ false, NamespaceBundleSplitAlgorithm.RANGE_EQUALLY_DIVIDE_NAME);
+
+ NamespaceBundles allBundlesAfterSplit =
+ pulsar1.getNamespaceService().getNamespaceBundleFactory()
+ .getBundles(namespaceName);
+
+ assertFalse(allBundlesAfterSplit.getBundles().contains(bundleWillBeSplit));
+
+ // the bundle data should be deleted
+
+ pulsar1.getBrokerService().updateRates();
+ pulsar2.getBrokerService().updateRates();
+
+ lm1.writeBrokerDataOnZooKeeper(true);
+ lm2.writeBrokerDataOnZooKeeper(true);
+
+ latch = new CountDownLatch(1);
+ // wait for metadata store notification finish
+ CountDownLatch finalLatch = latch;
+ executorService.submit(finalLatch::countDown);
+ latch.await();
+
+ loadManagerWrapper.writeResourceQuotasToZooKeeper();
+
+ lm1.updateAll();
+
+ log.info("update all triggered.");
+
+ // check bundle data should be deleted from metadata store.
+
+ CompletableFuture<List<String>> childrenAfterSplit = bundlesCache.getChildren(bundleDataPath);
+ List<String> bundlesAfterSplit = childrenAfterSplit.join();
+
+ assertFalse(bundlesAfterSplit.contains(bundleWillBeSplit.getBundleRange()));
+ }
+
}
[pulsar] 01/02: [fix][broker] fix bug caused by optimistic locking (#18390)
Posted by ba...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
baodi pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c2c9db3d6267dffc6b92422cf5cc503ac9b4fce6
Author: thetumbled <52...@users.noreply.github.com>
AuthorDate: Fri Sep 15 21:12:39 2023 +0800
[fix][broker] fix bug caused by optimistic locking (#18390)
(cherry picked from commit 2aa8c3b1a284673b512db2b815514343ec3e7c19)
---
.../util/collections/ConcurrentLongHashMap.java | 25 ++++-----
.../collections/ConcurrentLongLongPairHashMap.java | 39 ++++++++-----
.../util/collections/ConcurrentLongPairSet.java | 41 +++++++++-----
.../util/collections/ConcurrentOpenHashMap.java | 39 ++++++++-----
.../util/collections/ConcurrentOpenHashSet.java | 22 +++-----
.../apache/pulsar/common/util/FutureUtilTest.java | 1 -
.../collections/ConcurrentLongHashMapTest.java | 62 ++++++++++++++++++++-
.../ConcurrentLongLongPairHashMapTest.java | 65 ++++++++++++++++++++++
.../collections/ConcurrentLongPairSetTest.java | 64 +++++++++++++++++++++
.../collections/ConcurrentOpenHashMapTest.java | 64 +++++++++++++++++++++
.../collections/ConcurrentOpenHashSetTest.java | 64 +++++++++++++++++++++
11 files changed, 414 insertions(+), 72 deletions(-)
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMap.java
index 62303015c39..506a56d25ec 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMap.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMap.java
@@ -306,16 +306,17 @@ public class ConcurrentLongHashMap<V> {
}
V get(long key, int keyHash) {
- int bucket = keyHash;
-
long stamp = tryOptimisticRead();
boolean acquiredLock = false;
+ // add local variable here, so OutOfBound won't happen
+ long[] keys = this.keys;
+ V[] values = this.values;
+ // calculate table.length as capacity to avoid rehash changing capacity
+ int bucket = signSafeMod(keyHash, values.length);
+
try {
while (true) {
- int capacity = this.capacity;
- bucket = signSafeMod(bucket, capacity);
-
// First try optimistic locking
long storedKey = keys[bucket];
V storedValue = values[bucket];
@@ -333,16 +334,15 @@ public class ConcurrentLongHashMap<V> {
if (!acquiredLock) {
stamp = readLock();
acquiredLock = true;
+
+ // update local variable
+ keys = this.keys;
+ values = this.values;
+ bucket = signSafeMod(keyHash, values.length);
storedKey = keys[bucket];
storedValue = values[bucket];
}
- if (capacity != this.capacity) {
- // There has been a rehashing. We need to restart the search
- bucket = keyHash;
- continue;
- }
-
if (storedKey == key) {
return storedValue != DeletedValue ? storedValue : null;
} else if (storedValue == EmptyValue) {
@@ -350,8 +350,7 @@ public class ConcurrentLongHashMap<V> {
return null;
}
}
-
- ++bucket;
+ bucket = (bucket + 1) & (values.length - 1);
}
} finally {
if (acquiredLock) {
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java
index cce8be0b7fd..322c9bc160c 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java
@@ -284,6 +284,9 @@ public class ConcurrentLongLongPairHashMap {
// A section is a portion of the hash map that is covered by a single
@SuppressWarnings("serial")
private static final class Section extends StampedLock {
+ // Each item take up 4 continuous array space.
+ private static final int ITEM_SIZE = 4;
+
// Keys and values are stored interleaved in the table array
private volatile long[] table;
@@ -306,7 +309,7 @@ public class ConcurrentLongLongPairHashMap {
float expandFactor, float shrinkFactor) {
this.capacity = alignToPowerOfTwo(capacity);
this.initCapacity = this.capacity;
- this.table = new long[4 * this.capacity];
+ this.table = new long[ITEM_SIZE * this.capacity];
this.size = 0;
this.usedBuckets = 0;
this.autoShrink = autoShrink;
@@ -322,7 +325,10 @@ public class ConcurrentLongLongPairHashMap {
LongPair get(long key1, long key2, int keyHash) {
long stamp = tryOptimisticRead();
boolean acquiredLock = false;
- int bucket = signSafeMod(keyHash, capacity);
+ // add local variable here, so OutOfBound won't happen
+ long[] table = this.table;
+ // calculate table.length / 4 as capacity to avoid rehash changing capacity
+ int bucket = signSafeMod(keyHash, table.length / ITEM_SIZE);
try {
while (true) {
@@ -345,8 +351,9 @@ public class ConcurrentLongLongPairHashMap {
if (!acquiredLock) {
stamp = readLock();
acquiredLock = true;
-
- bucket = signSafeMod(keyHash, capacity);
+ // update local variable
+ table = this.table;
+ bucket = signSafeMod(keyHash, table.length / ITEM_SIZE);
storedKey1 = table[bucket];
storedKey2 = table[bucket + 1];
storedValue1 = table[bucket + 2];
@@ -361,7 +368,7 @@ public class ConcurrentLongLongPairHashMap {
}
}
- bucket = (bucket + 4) & (table.length - 1);
+ bucket = (bucket + ITEM_SIZE) & (table.length - 1);
}
} finally {
if (acquiredLock) {
@@ -413,7 +420,7 @@ public class ConcurrentLongLongPairHashMap {
}
}
- bucket = (bucket + 4) & (table.length - 1);
+ bucket = (bucket + ITEM_SIZE) & (table.length - 1);
}
} finally {
if (usedBuckets > resizeThresholdUp) {
@@ -454,7 +461,7 @@ public class ConcurrentLongLongPairHashMap {
return false;
}
- bucket = (bucket + 4) & (table.length - 1);
+ bucket = (bucket + ITEM_SIZE) & (table.length - 1);
}
} finally {
@@ -480,7 +487,7 @@ public class ConcurrentLongLongPairHashMap {
}
private void cleanBucket(int bucket) {
- int nextInArray = (bucket + 4) & (table.length - 1);
+ int nextInArray = (bucket + ITEM_SIZE) & (table.length - 1);
if (table[nextInArray] == EmptyKey) {
table[bucket] = EmptyKey;
table[bucket + 1] = EmptyKey;
@@ -489,7 +496,7 @@ public class ConcurrentLongLongPairHashMap {
--usedBuckets;
// Cleanup all the buckets that were in `DeletedKey` state, so that we can reduce unnecessary expansions
- bucket = (bucket - 4) & (table.length - 1);
+ bucket = (bucket - ITEM_SIZE) & (table.length - 1);
while (table[bucket] == DeletedKey) {
table[bucket] = EmptyKey;
table[bucket + 1] = EmptyKey;
@@ -497,7 +504,7 @@ public class ConcurrentLongLongPairHashMap {
table[bucket + 3] = ValueNotFound;
--usedBuckets;
- bucket = (bucket - 4) & (table.length - 1);
+ bucket = (bucket - ITEM_SIZE) & (table.length - 1);
}
} else {
table[bucket] = DeletedKey;
@@ -540,7 +547,7 @@ public class ConcurrentLongLongPairHashMap {
}
// Go through all the buckets for this section
- for (int bucket = 0; bucket < table.length; bucket += 4) {
+ for (int bucket = 0; bucket < table.length; bucket += ITEM_SIZE) {
long storedKey1 = table[bucket];
long storedKey2 = table[bucket + 1];
long storedValue1 = table[bucket + 2];
@@ -569,11 +576,11 @@ public class ConcurrentLongLongPairHashMap {
}
private void rehash(int newCapacity) {
- long[] newTable = new long[4 * newCapacity];
+ long[] newTable = new long[ITEM_SIZE * newCapacity];
Arrays.fill(newTable, EmptyKey);
// Re-hash table
- for (int i = 0; i < table.length; i += 4) {
+ for (int i = 0; i < table.length; i += ITEM_SIZE) {
long storedKey1 = table[i];
long storedKey2 = table[i + 1];
long storedValue1 = table[i + 2];
@@ -593,7 +600,7 @@ public class ConcurrentLongLongPairHashMap {
}
private void shrinkToInitCapacity() {
- long[] newTable = new long[4 * initCapacity];
+ long[] newTable = new long[ITEM_SIZE * initCapacity];
Arrays.fill(newTable, EmptyKey);
table = newTable;
@@ -622,7 +629,7 @@ public class ConcurrentLongLongPairHashMap {
return;
}
- bucket = (bucket + 4) & (table.length - 1);
+ bucket = (bucket + ITEM_SIZE) & (table.length - 1);
}
}
}
@@ -641,6 +648,8 @@ public class ConcurrentLongLongPairHashMap {
}
static final int signSafeMod(long n, int max) {
+ // as the ITEM_SIZE of Section is 4, so the index is the multiple of 4
+ // that is to left shift 2 bits
return (int) (n & (max - 1)) << 2;
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java
index 1a892dbd966..371595644c7 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java
@@ -294,6 +294,9 @@ public class ConcurrentLongPairSet implements LongPairSet {
// A section is a portion of the hash map that is covered by a single
@SuppressWarnings("serial")
private static final class Section extends StampedLock {
+ // Each item take up 2 continuous array space.
+ private static final int ITEM_SIZE = 2;
+
// Keys and values are stored interleaved in the table array
private volatile long[] table;
@@ -315,7 +318,7 @@ public class ConcurrentLongPairSet implements LongPairSet {
float expandFactor, float shrinkFactor) {
this.capacity = alignToPowerOfTwo(capacity);
this.initCapacity = this.capacity;
- this.table = new long[2 * this.capacity];
+ this.table = new long[ITEM_SIZE * this.capacity];
this.size = 0;
this.usedBuckets = 0;
this.autoShrink = autoShrink;
@@ -331,7 +334,11 @@ public class ConcurrentLongPairSet implements LongPairSet {
boolean contains(long item1, long item2, int hash) {
long stamp = tryOptimisticRead();
boolean acquiredLock = false;
- int bucket = signSafeMod(hash, capacity);
+
+ // add local variable here, so OutOfBound won't happen
+ long[] table = this.table;
+ // calculate table.length / 2 as capacity to avoid rehash changing capacity
+ int bucket = signSafeMod(hash, table.length / ITEM_SIZE);
try {
while (true) {
@@ -353,7 +360,9 @@ public class ConcurrentLongPairSet implements LongPairSet {
stamp = readLock();
acquiredLock = true;
- bucket = signSafeMod(hash, capacity);
+ // update local variable
+ table = this.table;
+ bucket = signSafeMod(hash, table.length / ITEM_SIZE);
storedItem1 = table[bucket];
storedItem2 = table[bucket + 1];
}
@@ -366,7 +375,7 @@ public class ConcurrentLongPairSet implements LongPairSet {
}
}
- bucket = (bucket + 2) & (table.length - 1);
+ bucket = (bucket + ITEM_SIZE) & (table.length - 1);
}
} finally {
if (acquiredLock) {
@@ -410,7 +419,7 @@ public class ConcurrentLongPairSet implements LongPairSet {
}
}
- bucket = (bucket + 2) & (table.length - 1);
+ bucket = (bucket + ITEM_SIZE) & (table.length - 1);
}
} finally {
if (usedBuckets > resizeThresholdUp) {
@@ -445,7 +454,7 @@ public class ConcurrentLongPairSet implements LongPairSet {
return false;
}
- bucket = (bucket + 2) & (table.length - 1);
+ bucket = (bucket + ITEM_SIZE) & (table.length - 1);
}
} finally {
tryShrinkThenUnlock(stamp);
@@ -459,7 +468,7 @@ public class ConcurrentLongPairSet implements LongPairSet {
// Go through all the buckets for this section
long stamp = writeLock();
try {
- for (int bucket = 0; bucket < table.length; bucket += 2) {
+ for (int bucket = 0; bucket < table.length; bucket += ITEM_SIZE) {
long storedItem1 = table[bucket];
long storedItem2 = table[bucket + 1];
if (storedItem1 != DeletedItem && storedItem1 != EmptyItem) {
@@ -498,7 +507,7 @@ public class ConcurrentLongPairSet implements LongPairSet {
}
private void cleanBucket(int bucket) {
- int nextInArray = (bucket + 2) & (table.length - 1);
+ int nextInArray = (bucket + ITEM_SIZE) & (table.length - 1);
if (table[nextInArray] == EmptyItem) {
table[bucket] = EmptyItem;
table[bucket + 1] = EmptyItem;
@@ -506,13 +515,13 @@ public class ConcurrentLongPairSet implements LongPairSet {
// Cleanup all the buckets that were in `DeletedItem` state,
// so that we can reduce unnecessary expansions
- int lastBucket = (bucket - 2) & (table.length - 1);
+ int lastBucket = (bucket - ITEM_SIZE) & (table.length - 1);
while (table[lastBucket] == DeletedItem) {
table[lastBucket] = EmptyItem;
table[lastBucket + 1] = EmptyItem;
--usedBuckets;
- lastBucket = (lastBucket - 2) & (table.length - 1);
+ lastBucket = (lastBucket - ITEM_SIZE) & (table.length - 1);
}
} else {
table[bucket] = DeletedItem;
@@ -542,7 +551,7 @@ public class ConcurrentLongPairSet implements LongPairSet {
// Go through all the buckets for this section. We try to renew the stamp only after a validation
// error, otherwise we keep going with the same.
long stamp = 0;
- for (int bucket = 0; bucket < table.length; bucket += 2) {
+ for (int bucket = 0; bucket < table.length; bucket += ITEM_SIZE) {
if (stamp == 0) {
stamp = tryOptimisticRead();
}
@@ -572,11 +581,11 @@ public class ConcurrentLongPairSet implements LongPairSet {
private void rehash(int newCapacity) {
// Expand the hashmap
- long[] newTable = new long[2 * newCapacity];
+ long[] newTable = new long[ITEM_SIZE * newCapacity];
Arrays.fill(newTable, EmptyItem);
// Re-hash table
- for (int i = 0; i < table.length; i += 2) {
+ for (int i = 0; i < table.length; i += ITEM_SIZE) {
long storedItem1 = table[i];
long storedItem2 = table[i + 1];
if (storedItem1 != EmptyItem && storedItem1 != DeletedItem) {
@@ -595,7 +604,7 @@ public class ConcurrentLongPairSet implements LongPairSet {
private void shrinkToInitCapacity() {
// Expand the hashmap
- long[] newTable = new long[2 * initCapacity];
+ long[] newTable = new long[ITEM_SIZE * initCapacity];
Arrays.fill(newTable, EmptyItem);
table = newTable;
@@ -621,7 +630,7 @@ public class ConcurrentLongPairSet implements LongPairSet {
return;
}
- bucket = (bucket + 2) & (table.length - 1);
+ bucket = (bucket + ITEM_SIZE) & (table.length - 1);
}
}
}
@@ -640,6 +649,8 @@ public class ConcurrentLongPairSet implements LongPairSet {
}
static final int signSafeMod(long n, int max) {
+ // as the ITEM_SIZE of Section is 2, so the index is the multiple of 2
+ // that is to left shift 1 bit
return (int) (n & (max - 1)) << 1;
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java
index 618b3e539a9..fc39bcf6374 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java
@@ -291,6 +291,9 @@ public class ConcurrentOpenHashMap<K, V> {
// A section is a portion of the hash map that is covered by a single
@SuppressWarnings("serial")
private static final class Section<K, V> extends StampedLock {
+ // Each item take up 2 continuous array space.
+ private static final int ITEM_SIZE = 2;
+
// Keys and values are stored interleaved in the table array
private volatile Object[] table;
@@ -312,7 +315,7 @@ public class ConcurrentOpenHashMap<K, V> {
float expandFactor, float shrinkFactor) {
this.capacity = alignToPowerOfTwo(capacity);
this.initCapacity = this.capacity;
- this.table = new Object[2 * this.capacity];
+ this.table = new Object[ITEM_SIZE * this.capacity];
this.size = 0;
this.usedBuckets = 0;
this.autoShrink = autoShrink;
@@ -327,7 +330,11 @@ public class ConcurrentOpenHashMap<K, V> {
V get(K key, int keyHash) {
long stamp = tryOptimisticRead();
boolean acquiredLock = false;
- int bucket = signSafeMod(keyHash, capacity);
+
+ // add local variable here, so OutOfBound won't happen
+ Object[] table = this.table;
+ // calculate table.length / 2 as capacity to avoid rehash changing capacity
+ int bucket = signSafeMod(keyHash, table.length / ITEM_SIZE);
try {
while (true) {
@@ -349,7 +356,9 @@ public class ConcurrentOpenHashMap<K, V> {
stamp = readLock();
acquiredLock = true;
- bucket = signSafeMod(keyHash, capacity);
+ // update local variable
+ table = this.table;
+ bucket = signSafeMod(keyHash, table.length / ITEM_SIZE);
storedKey = (K) table[bucket];
storedValue = (V) table[bucket + 1];
}
@@ -362,7 +371,7 @@ public class ConcurrentOpenHashMap<K, V> {
}
}
- bucket = (bucket + 2) & (table.length - 1);
+ bucket = (bucket + ITEM_SIZE) & (table.length - 1);
}
} finally {
if (acquiredLock) {
@@ -415,7 +424,7 @@ public class ConcurrentOpenHashMap<K, V> {
}
}
- bucket = (bucket + 2) & (table.length - 1);
+ bucket = (bucket + ITEM_SIZE) & (table.length - 1);
}
} finally {
if (usedBuckets > resizeThresholdUp) {
@@ -444,7 +453,7 @@ public class ConcurrentOpenHashMap<K, V> {
if (value == null || value.equals(storedValue)) {
SIZE_UPDATER.decrementAndGet(this);
- int nextInArray = (bucket + 2) & (table.length - 1);
+ int nextInArray = (bucket + ITEM_SIZE) & (table.length - 1);
if (table[nextInArray] == EmptyKey) {
table[bucket] = EmptyKey;
table[bucket + 1] = null;
@@ -452,13 +461,13 @@ public class ConcurrentOpenHashMap<K, V> {
// Cleanup all the buckets that were in `DeletedKey` state,
// so that we can reduce unnecessary expansions
- int lastBucket = (bucket - 2) & (table.length - 1);
+ int lastBucket = (bucket - ITEM_SIZE) & (table.length - 1);
while (table[lastBucket] == DeletedKey) {
table[lastBucket] = EmptyKey;
table[lastBucket + 1] = null;
--usedBuckets;
- lastBucket = (lastBucket - 2) & (table.length - 1);
+ lastBucket = (lastBucket - ITEM_SIZE) & (table.length - 1);
}
} else {
table[bucket] = DeletedKey;
@@ -474,7 +483,7 @@ public class ConcurrentOpenHashMap<K, V> {
return null;
}
- bucket = (bucket + 2) & (table.length - 1);
+ bucket = (bucket + ITEM_SIZE) & (table.length - 1);
}
} finally {
@@ -523,7 +532,7 @@ public class ConcurrentOpenHashMap<K, V> {
// Go through all the buckets for this section. We try to renew the stamp only after a validation
// error, otherwise we keep going with the same.
long stamp = 0;
- for (int bucket = 0; bucket < table.length; bucket += 2) {
+ for (int bucket = 0; bucket < table.length; bucket += ITEM_SIZE) {
if (stamp == 0) {
stamp = tryOptimisticRead();
}
@@ -553,10 +562,10 @@ public class ConcurrentOpenHashMap<K, V> {
private void rehash(int newCapacity) {
// Expand the hashmap
- Object[] newTable = new Object[2 * newCapacity];
+ Object[] newTable = new Object[ITEM_SIZE * newCapacity];
// Re-hash table
- for (int i = 0; i < table.length; i += 2) {
+ for (int i = 0; i < table.length; i += ITEM_SIZE) {
K storedKey = (K) table[i];
V storedValue = (V) table[i + 1];
if (storedKey != EmptyKey && storedKey != DeletedKey) {
@@ -572,7 +581,7 @@ public class ConcurrentOpenHashMap<K, V> {
}
private void shrinkToInitCapacity() {
- Object[] newTable = new Object[2 * initCapacity];
+ Object[] newTable = new Object[ITEM_SIZE * initCapacity];
table = newTable;
size = 0;
@@ -597,7 +606,7 @@ public class ConcurrentOpenHashMap<K, V> {
return;
}
- bucket = (bucket + 2) & (table.length - 1);
+ bucket = (bucket + ITEM_SIZE) & (table.length - 1);
}
}
}
@@ -613,6 +622,8 @@ public class ConcurrentOpenHashMap<K, V> {
}
static final int signSafeMod(long n, int max) {
+ // as the ITEM_SIZE of Section is 2, so the index is the multiple of 2
+ // that is to left shift 1 bit
return (int) (n & (max - 1)) << 1;
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSet.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSet.java
index a5333adae2f..51945e42c0a 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSet.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSet.java
@@ -294,16 +294,16 @@ public class ConcurrentOpenHashSet<V> {
}
boolean contains(V value, int keyHash) {
- int bucket = keyHash;
-
long stamp = tryOptimisticRead();
boolean acquiredLock = false;
+ // add local variable here, so OutOfBound won't happen
+ V[] values = this.values;
+ // calculate table.length as capacity to avoid rehash changing capacity
+ int bucket = signSafeMod(keyHash, values.length);
+
try {
while (true) {
- int capacity = this.capacity;
- bucket = signSafeMod(bucket, capacity);
-
// First try optimistic locking
V storedValue = values[bucket];
@@ -321,15 +321,12 @@ public class ConcurrentOpenHashSet<V> {
stamp = readLock();
acquiredLock = true;
+ // update local variable
+ values = this.values;
+ bucket = signSafeMod(keyHash, values.length);
storedValue = values[bucket];
}
- if (capacity != this.capacity) {
- // There has been a rehashing. We need to restart the search
- bucket = keyHash;
- continue;
- }
-
if (value.equals(storedValue)) {
return true;
} else if (storedValue == EmptyValue) {
@@ -337,8 +334,7 @@ public class ConcurrentOpenHashSet<V> {
return false;
}
}
-
- ++bucket;
+ bucket = (bucket + 1) & (values.length - 1);
}
} finally {
if (acquiredLock) {
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/FutureUtilTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/FutureUtilTest.java
index 3899f244512..83cd856bff5 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/FutureUtilTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/FutureUtilTest.java
@@ -182,7 +182,6 @@ public class FutureUtilTest {
}
}
- @Test
public void testSequencer() {
int concurrentNum = 1000;
final ScheduledExecutorService executor = Executors.newScheduledThreadPool(concurrentNum);
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java
index f60fea7ab7f..71c68811ae1 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java
@@ -38,6 +38,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.LongFunction;
import lombok.Cleanup;
@@ -213,6 +214,66 @@ public class ConcurrentLongHashMapTest {
assertTrue(map.capacity() == initCapacity);
}
+ @Test
+ public void testConcurrentExpandAndShrinkAndGet() throws Throwable {
+ ConcurrentLongHashMap<String> map = ConcurrentLongHashMap.<String>newBuilder()
+ .expectedItems(2)
+ .concurrencyLevel(1)
+ .autoShrink(true)
+ .mapIdleFactor(0.25f)
+ .build();
+ assertEquals(map.capacity(), 4);
+
+ ExecutorService executor = Executors.newCachedThreadPool();
+ final int readThreads = 16;
+ final int writeThreads = 1;
+ final int n = 1_000;
+ CyclicBarrier barrier = new CyclicBarrier(writeThreads + readThreads);
+ Future<?> future = null;
+ AtomicReference<Exception> ex = new AtomicReference<>();
+
+ for (int i = 0; i < readThreads; i++) {
+ executor.submit(() -> {
+ try {
+ barrier.await();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ try {
+ map.get(1);
+ } catch (Exception e) {
+ ex.set(e);
+ }
+ });
+ }
+
+ assertNull(map.put(1,"v1"));
+ future = executor.submit(() -> {
+ try {
+ barrier.await();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ for (int i = 0; i < n; i++) {
+ // expand hashmap
+ assertNull(map.put(2, "v2"));
+ assertNull(map.put(3, "v3"));
+ assertEquals(map.capacity(), 8);
+
+ // shrink hashmap
+ assertTrue(map.remove(2, "v2"));
+ assertTrue(map.remove(3, "v3"));
+ assertEquals(map.capacity(), 4);
+ }
+ });
+
+ future.get();
+ assertTrue(ex.get() == null);
+ // shut down pool
+ executor.shutdown();
+ }
+
@Test
public void testRemove() {
ConcurrentLongHashMap<String> map = ConcurrentLongHashMap.<String>newBuilder()
@@ -361,7 +422,6 @@ public class ConcurrentLongHashMapTest {
assertEquals(map.size(), N * nThreads);
}
- @Test
public void stressConcurrentInsertionsAndReads() throws Throwable {
ConcurrentLongHashMap<String> map = ConcurrentLongHashMap.<String>newBuilder()
.expectedItems(4)
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java
index eca63734915..b561d61a1ff 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java
@@ -31,9 +31,12 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
+import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicReference;
+
import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap.LongPair;
import org.testng.annotations.Test;
@@ -173,6 +176,68 @@ public class ConcurrentLongLongPairHashMapTest {
assertEquals(map.capacity(), 8);
}
+ @Test
+ public void testConcurrentExpandAndShrinkAndGet() throws Throwable {
+ ConcurrentLongLongPairHashMap map = ConcurrentLongLongPairHashMap.newBuilder()
+ .expectedItems(2)
+ .concurrencyLevel(1)
+ .autoShrink(true)
+ .mapIdleFactor(0.25f)
+ .build();
+ assertEquals(map.capacity(), 4);
+
+ ExecutorService executor = Executors.newCachedThreadPool();
+ final int readThreads = 16;
+ final int writeThreads = 1;
+ final int n = 1_000;
+ CyclicBarrier barrier = new CyclicBarrier(writeThreads + readThreads);
+ Future<?> future = null;
+ AtomicReference<Exception> ex = new AtomicReference<>();
+
+ for (int i = 0; i < readThreads; i++) {
+ executor.submit(() -> {
+ try {
+ barrier.await();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ while (true) {
+ try {
+ map.get(1, 1);
+ } catch (Exception e) {
+ ex.set(e);
+ }
+ }
+ });
+ }
+
+ assertTrue(map.put(1, 1, 11, 11));
+ future = executor.submit(() -> {
+ try {
+ barrier.await();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ for (int i = 0; i < n; i++) {
+ // expand hashmap
+ assertTrue(map.put(2, 2, 22, 22));
+ assertTrue(map.put(3, 3, 33, 33));
+ assertEquals(map.capacity(), 8);
+
+ // shrink hashmap
+ assertTrue(map.remove(2, 2, 22, 22));
+ assertTrue(map.remove(3, 3, 33, 33));
+ assertEquals(map.capacity(), 4);
+ }
+ });
+
+ future.get();
+ assertTrue(ex.get() == null);
+ // shut down pool
+ executor.shutdown();
+ }
+
@Test
public void testExpandShrinkAndClear() {
ConcurrentLongLongPairHashMap map = ConcurrentLongLongPairHashMap.newBuilder()
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSetTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSetTest.java
index a7bbe9b0dc3..9863cde4c97 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSetTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSetTest.java
@@ -30,9 +30,11 @@ import java.util.HashSet;
import java.util.List;
import java.util.Random;
import java.util.Set;
+import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicReference;
import lombok.Cleanup;
import org.apache.pulsar.common.util.collections.ConcurrentLongPairSet.LongPair;
@@ -211,6 +213,68 @@ public class ConcurrentLongPairSetTest {
assertTrue(map.capacity() == initCapacity);
}
+ @Test
+ public void testConcurrentExpandAndShrinkAndGet() throws Throwable {
+ ConcurrentLongPairSet set = ConcurrentLongPairSet.newBuilder()
+ .expectedItems(2)
+ .concurrencyLevel(1)
+ .autoShrink(true)
+ .mapIdleFactor(0.25f)
+ .build();
+ assertEquals(set.capacity(), 4);
+
+ ExecutorService executor = Executors.newCachedThreadPool();
+ final int readThreads = 16;
+ final int writeThreads = 1;
+ final int n = 1_000;
+ CyclicBarrier barrier = new CyclicBarrier(writeThreads + readThreads);
+ Future<?> future = null;
+ AtomicReference<Exception> ex = new AtomicReference<>();
+
+ for (int i = 0; i < readThreads; i++) {
+ executor.submit(() -> {
+ try {
+ barrier.await();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ while (true) {
+ try {
+ set.contains(1, 1);
+ } catch (Exception e) {
+ ex.set(e);
+ }
+ }
+ });
+ }
+
+ assertTrue(set.add(1, 1));
+ future = executor.submit(() -> {
+ try {
+ barrier.await();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ for (int i = 0; i < n; i++) {
+ // expand hashmap
+ assertTrue(set.add(2, 2));
+ assertTrue(set.add(3, 3));
+ assertEquals(set.capacity(), 8);
+
+ // shrink hashmap
+ assertTrue(set.remove(2, 2));
+ assertTrue(set.remove(3, 3));
+ assertEquals(set.capacity(), 4);
+ }
+ });
+
+ future.get();
+ assertTrue(ex.get() == null);
+ // shut down pool
+ executor.shutdown();
+ }
+
@Test
public void testRemove() {
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java
index 411dfc735fd..de852797329 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java
@@ -32,11 +32,13 @@ import java.util.HashMap;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import lombok.Cleanup;
@@ -215,6 +217,68 @@ public class ConcurrentOpenHashMapTest {
assertTrue(map.capacity() == initCapacity);
}
+ @Test
+ public void testConcurrentExpandAndShrinkAndGet() throws Throwable {
+ ConcurrentOpenHashMap<String, String> map = ConcurrentOpenHashMap.<String, String>newBuilder()
+ .expectedItems(2)
+ .concurrencyLevel(1)
+ .autoShrink(true)
+ .mapIdleFactor(0.25f)
+ .build();
+ assertEquals(map.capacity(), 4);
+
+ ExecutorService executor = Executors.newCachedThreadPool();
+ final int readThreads = 16;
+ final int writeThreads = 1;
+ final int n = 1_000;
+ CyclicBarrier barrier = new CyclicBarrier(writeThreads + readThreads);
+ Future<?> future = null;
+ AtomicReference<Exception> ex = new AtomicReference<>();
+
+ for (int i = 0; i < readThreads; i++) {
+ executor.submit(() -> {
+ try {
+ barrier.await();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ while (true) {
+ try {
+ map.get("k2");
+ } catch (Exception e) {
+ ex.set(e);
+ }
+ }
+ });
+ }
+
+ assertNull(map.put("k1","v1"));
+ future = executor.submit(() -> {
+ try {
+ barrier.await();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ for (int i = 0; i < n; i++) {
+ // expand hashmap
+ assertNull(map.put("k2", "v2"));
+ assertNull(map.put("k3", "v3"));
+ assertEquals(map.capacity(), 8);
+
+ // shrink hashmap
+ assertTrue(map.remove("k2", "v2"));
+ assertTrue(map.remove("k3", "v3"));
+ assertEquals(map.capacity(), 4);
+ }
+ });
+
+ future.get();
+ assertTrue(ex.get() == null);
+ // shut down pool
+ executor.shutdown();
+ }
+
@Test
public void testRemove() {
ConcurrentOpenHashMap<String, String> map =
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java
index d6403af52cf..37cd4df0660 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java
@@ -28,9 +28,11 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Random;
+import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicReference;
import lombok.Cleanup;
import org.testng.annotations.Test;
@@ -186,6 +188,68 @@ public class ConcurrentOpenHashSetTest {
assertTrue(map.capacity() == initCapacity);
}
+ @Test
+ public void testConcurrentExpandAndShrinkAndGet() throws Throwable {
+ ConcurrentOpenHashSet<String> set = ConcurrentOpenHashSet.<String>newBuilder()
+ .expectedItems(2)
+ .concurrencyLevel(1)
+ .autoShrink(true)
+ .mapIdleFactor(0.25f)
+ .build();
+ assertEquals(set.capacity(), 4);
+
+ ExecutorService executor = Executors.newCachedThreadPool();
+ final int readThreads = 16;
+ final int writeThreads = 1;
+ final int n = 1_000;
+ CyclicBarrier barrier = new CyclicBarrier(writeThreads + readThreads);
+ Future<?> future = null;
+ AtomicReference<Exception> ex = new AtomicReference<>();
+
+ for (int i = 0; i < readThreads; i++) {
+ executor.submit(() -> {
+ try {
+ barrier.await();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ while (true) {
+ try {
+ set.contains("k2");
+ } catch (Exception e) {
+ ex.set(e);
+ }
+ }
+ });
+ }
+
+ assertTrue(set.add("k1"));
+ future = executor.submit(() -> {
+ try {
+ barrier.await();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ for (int i = 0; i < n; i++) {
+ // expand hashmap
+ assertTrue(set.add("k2"));
+ assertTrue(set.add("k3"));
+ assertEquals(set.capacity(), 8);
+
+ // shrink hashmap
+ assertTrue(set.remove("k2"));
+ assertTrue(set.remove("k3"));
+ assertEquals(set.capacity(), 4);
+ }
+ });
+
+ future.get();
+ assertTrue(ex.get() == null);
+ // shut down pool
+ executor.shutdown();
+ }
+
@Test
public void testRemove() {
ConcurrentOpenHashSet<String> set =