[activemq-artemis] branch main updated: ARTEMIS-4417: use ConcurrentHashMap for the sub-maps to make their use thread safe

[brusdev]

This is an automated email from the ASF dual-hosted git repository.

brusdev pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git


The following commit(s) were added to refs/heads/main by this push:
     new aa4e4409bf ARTEMIS-4417: use ConcurrentHashMap for the sub-maps to 
make their use thread safe
aa4e4409bf is described below

commit aa4e4409bf2cf4b14a02b5b2887b900e30cae84d
Author: Robbie Gemmell <rob...@apache.org>
AuthorDate: Tue Sep 5 13:58:19 2023 +0100

    ARTEMIS-4417: use ConcurrentHashMap for the sub-maps to make their use 
thread safe
---
 .../journal/AbstractJournalStorageManager.java     | 29 ++++++++++++----------
 1 file changed, 16 insertions(+), 13 deletions(-)

diff --git 
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
 
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
index 9a4d8533f0..23800a5577 100644
--- 
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
+++ 
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
@@ -33,6 +33,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
@@ -232,7 +233,7 @@ public abstract class AbstractJournalStorageManager extends 
CriticalComponentImp
 
    protected final Map<String, PersistedRole> mapPersistedRoles = new 
ConcurrentHashMap<>();
 
-   protected final Map<String, Map<String, PersistedKeyValuePair>> 
mapPersistedKeyValuePairs = new ConcurrentHashMap<>();
+   protected final ConcurrentMap<String, ConcurrentMap<String, 
PersistedKeyValuePair>> mapPersistedKeyValuePairs = new ConcurrentHashMap<>();
 
    protected final ConcurrentLongHashMap<LargeServerMessage> 
largeMessagesToDelete = new ConcurrentLongHashMap<>();
 
@@ -891,12 +892,7 @@ public abstract class AbstractJournalStorageManager 
extends CriticalComponentImp
          final long id = idGenerator.generateID();
          persistedKeyValuePair.setStoreId(id);
          bindingsJournal.appendAddRecord(id, 
JournalRecordIds.KEY_VALUE_PAIR_RECORD, persistedKeyValuePair, true);
-         Map<String, PersistedKeyValuePair> persistedKeyValuePairs = 
mapPersistedKeyValuePairs.get(persistedKeyValuePair.getMapId());
-         if (persistedKeyValuePairs == null) {
-            persistedKeyValuePairs = new HashMap<>();
-            mapPersistedKeyValuePairs.put(persistedKeyValuePair.getMapId(), 
persistedKeyValuePairs);
-         }
-         persistedKeyValuePairs.put(persistedKeyValuePair.getKey(), 
persistedKeyValuePair);
+         insertPersistedKeyValuePair(persistedKeyValuePair);
       }
    }
 
@@ -1658,12 +1654,7 @@ public abstract class AbstractJournalStorageManager 
extends CriticalComponentImp
                mapPersistedRoles.put(role.getUsername(), role);
             } else if (rec == JournalRecordIds.KEY_VALUE_PAIR_RECORD) {
                PersistedKeyValuePair keyValuePair = 
newKeyValuePairEncoding(id, buffer);
-               Map<String, PersistedKeyValuePair> persistedKeyValuePairs = 
mapPersistedKeyValuePairs.get(keyValuePair.getMapId());
-               if (persistedKeyValuePairs == null) {
-                  persistedKeyValuePairs = new HashMap<>();
-                  mapPersistedKeyValuePairs.put(keyValuePair.getMapId(), 
persistedKeyValuePairs);
-               }
-               persistedKeyValuePairs.put(keyValuePair.getKey(), keyValuePair);
+               insertPersistedKeyValuePair(keyValuePair);
             } else if (rec == JournalRecordIds.CONNECTOR_RECORD) {
                PersistedConnector connector = newConnectorEncoding(id, buffer);
                mapPersistedConnectors.put(connector.getName(), connector);
@@ -1690,6 +1681,18 @@ public abstract class AbstractJournalStorageManager 
extends CriticalComponentImp
       return bindingsInfo;
    }
 
+   private void insertPersistedKeyValuePair(final PersistedKeyValuePair 
keyValuePair) {
+      Map<String, PersistedKeyValuePair> persistedKeyValuePairs = 
mapPersistedKeyValuePairs.get(keyValuePair.getMapId());
+      if (persistedKeyValuePairs == null) {
+         ConcurrentMap<String, PersistedKeyValuePair> newMap = new 
ConcurrentHashMap<>();
+         Map<String, PersistedKeyValuePair> existingMap = 
mapPersistedKeyValuePairs.putIfAbsent(keyValuePair.getMapId(), newMap);
+
+         persistedKeyValuePairs = existingMap == null ? newMap : existingMap;
+      }
+
+      persistedKeyValuePairs.put(keyValuePair.getKey(), keyValuePair);
+   }
+
    @Override
    public void lineUpContext() {
       try (ArtemisCloseable lock = closeableReadLock()) {