Fix PersistenceQuery ReadJournalFor thread safety issue (#6859)

akkadotnet · Jul 31, 2023 · 0c43702 · 0c43702
1 parent 7ee2dbb
commit 0c43702
Showing 1 changed file with 13 additions and 2 deletions.
diff --git a/src/core/Akka.Persistence.Query/PersistenceQuery.cs b/src/core/Akka.Persistence.Query/PersistenceQuery.cs
@@ -19,6 +19,7 @@ public sealed class PersistenceQuery : IExtension
         private readonly ExtendedActorSystem _system;
         private readonly ConcurrentDictionary<string, IReadJournal> _readJournalPluginExtensionIds = new();
         private ILoggingAdapter _log;
+        private readonly object _lock = new ();
 
         public static PersistenceQuery Get(ActorSystem system)
         {
@@ -34,8 +35,18 @@ public PersistenceQuery(ExtendedActorSystem system)
 
         public TJournal ReadJournalFor<TJournal>(string readJournalPluginId) where TJournal : IReadJournal
         {
-            var plugin = _readJournalPluginExtensionIds.GetOrAdd(readJournalPluginId, path => CreatePlugin(path, GetDefaultConfig<TJournal>()).GetReadJournal());
-            return (TJournal)plugin;
+            if(_readJournalPluginExtensionIds.TryGetValue(readJournalPluginId, out var plugin))
+                return (TJournal)plugin;
+
+            lock (_lock)
+            {
+                if (_readJournalPluginExtensionIds.TryGetValue(readJournalPluginId, out plugin))
+                    return (TJournal)plugin;
+
+                plugin = CreatePlugin(readJournalPluginId, GetDefaultConfig<TJournal>()).GetReadJournal();
+                _readJournalPluginExtensionIds[readJournalPluginId] = plugin;
+                return (TJournal)plugin;
+            }
         }
 
         private IReadJournalProvider CreatePlugin(string configPath, Config config)