(kafka) branch trunk updated: KAFKA-16055: Thread unsafe use of HashMap stored in QueryableStoreProvider#storeProviders (#15121)

[ab...@apache.org]

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

ableegoldman pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 87c7fc0cd73 KAFKA-16055: Thread unsafe use of HashMap stored in QueryableStoreProvider#storeProviders (#15121)
87c7fc0cd73 is described below

commit 87c7fc0cd73be49c7d942ad23caf266d5b71263c
Author: Kohei Nozaki <ko...@apache.org>
AuthorDate: Thu Feb 8 06:02:42 2024 +0900

    KAFKA-16055: Thread unsafe use of HashMap stored in QueryableStoreProvider#storeProviders (#15121)
    
    This PR replaces a HashMap by a ConcurrentHashMap so that the local state store queries can be made from multiple threads. See this for additional context: https://lists.apache.org/thread/gpct1275bfqovlckptn3lvf683qpoxol
    
    Reviewers: Anna Sophie Blee-Goldman <ab...@apache.org>, Sagar Rao <sa...@gmail.com>
---
 .../apache/kafka/streams/state/internals/QueryableStoreProvider.java  | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/QueryableStoreProvider.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/QueryableStoreProvider.java
index 07cf0ee8607..7ed11917b9c 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/QueryableStoreProvider.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/QueryableStoreProvider.java
@@ -20,7 +20,7 @@ import org.apache.kafka.streams.StoreQueryParameters;
 import org.apache.kafka.streams.processor.StateStore;
 import org.apache.kafka.streams.state.QueryableStoreType;
 
-import java.util.HashMap;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -36,7 +36,7 @@ public class QueryableStoreProvider {
     private final GlobalStateStoreProvider globalStoreProvider;
 
     public QueryableStoreProvider(final GlobalStateStoreProvider globalStateStoreProvider) {
-        this.storeProviders = new HashMap<>();
+        this.storeProviders = new ConcurrentHashMap<>();
         this.globalStoreProvider = globalStateStoreProvider;
     }