[FLINK-37166] Use concurrenthashmap in flink config manager

Author: gyfora

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java

@@ -50,13 +50,13 @@
 
 import java.time.Duration;
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.SortedMap;
 import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -87,7 +87,7 @@ public class FlinkConfigManager {
     private final AtomicLong defaultConfigVersion = new AtomicLong(0);
     private final LoadingCache<Key, Configuration> cache;
     private final Consumer<Set<String>> namespaceListener;
-    private volatile Map<FlinkVersion, List<String>> relevantFlinkVersionPrefixes;
+    private volatile ConcurrentHashMap<FlinkVersion, List<String>> relevantFlinkVersionPrefixes;
 
     protected static final Pattern FLINK_VERSION_PATTERN =
             Pattern.compile(
@@ -114,7 +114,7 @@ public FlinkConfigManager(
         this.namespaceListener = namespaceListener;
         Duration cacheTimeout =
                 defaultConfig.get(KubernetesOperatorConfigOptions.OPERATOR_CONFIG_CACHE_TIMEOUT);
-        this.relevantFlinkVersionPrefixes = new HashMap<>();
+        this.relevantFlinkVersionPrefixes = new ConcurrentHashMap<>();
         this.cache =
                 CacheBuilder.newBuilder()
                         .maximumSize(
@@ -189,7 +189,7 @@ public void updateDefaultConfig(Configuration newConf) {
         // We clear the cached relevant Flink version prefixes as the base config may include new
         // version overrides.
         // This will trigger a regeneration of the prefixes in the next call to getDefaultConfig.
-        relevantFlinkVersionPrefixes = new HashMap<>();
+        relevantFlinkVersionPrefixes = new ConcurrentHashMap<>();
     }
 
     /**