keep synchronized for regions

Signed-off-by: PengFei Li <[email protected]>
banmoy · Aug 23, 2023 · 96c9158 · 96c9158
1 parent 3314894
commit 96c9158
Showing 1 changed file with 12 additions and 5 deletions.
diff --git a/...-stream-load-sdk/src/main/java/com/starrocks/data/load/stream/v2/StreamLoadManagerV2.java b/...-stream-load-sdk/src/main/java/com/starrocks/data/load/stream/v2/StreamLoadManagerV2.java
@@ -419,11 +419,18 @@ protected TableRegion getCacheRegion(String uniqueKey, String database, String t
 
         TableRegion region = regions.get(uniqueKey);
         if (region == null) {
-            StreamLoadTableProperties tableProperties = properties.getTableProperties(uniqueKey);
-            region = new TransactionTableRegion(uniqueKey, database, table, this,
-                    tableProperties, streamLoader, maxRetries, retryIntervalInMs);
-            regions.put(uniqueKey, region);
-            flushQ.offer((TransactionTableRegion) region);
+            // currently write() will not be called concurrently, so regions will also not be
+            // created concurrently, but for future extension, protect it with synchronized
+            synchronized (regions) {
+                region = regions.get(uniqueKey);
+                if (region == null) {
+                    StreamLoadTableProperties tableProperties = properties.getTableProperties(uniqueKey);
+                    region = new TransactionTableRegion(uniqueKey, database, table, this,
+                            tableProperties, streamLoader, maxRetries, retryIntervalInMs);
+                    regions.put(uniqueKey, region);
+                    flushQ.offer((TransactionTableRegion) region);
+                }
+            }
         }
         return region;
     }