[Dataflow Streaming] Increase windmill state cache concurrency level (#…

…33688)
apache · Jan 22, 2025 · c761112 · c761112
1 parent cdad92b
commit c761112
Showing 1 changed file with 4 additions and 2 deletions.
diff --git a/.../main/java/org/apache/beam/runners/dataflow/worker/windmill/state/WindmillStateCache.java b/.../main/java/org/apache/beam/runners/dataflow/worker/windmill/state/WindmillStateCache.java
@@ -79,15 +79,17 @@ public class WindmillStateCache implements StatusDataProvider {
 
   WindmillStateCache(long sizeMb, boolean supportMapViaMultimap) {
     this.workerCacheBytes = sizeMb * MEGABYTES;
+    int stateCacheConcurrencyLevel =
+        Math.max(STATE_CACHE_CONCURRENCY_LEVEL, Runtime.getRuntime().availableProcessors());
     this.stateCache =
         CacheBuilder.newBuilder()
             .maximumWeight(workerCacheBytes)
             .recordStats()
             .weigher(Weighers.weightedKeysAndValues())
-            .concurrencyLevel(STATE_CACHE_CONCURRENCY_LEVEL)
+            .concurrencyLevel(stateCacheConcurrencyLevel)
             .build();
     this.keyIndex =
-        new MapMaker().weakValues().concurrencyLevel(STATE_CACHE_CONCURRENCY_LEVEL).makeMap();
+        new MapMaker().weakValues().concurrencyLevel(stateCacheConcurrencyLevel).makeMap();
     this.supportMapViaMultimap = supportMapViaMultimap;
   }