[Bugfix] Fix NPE caused by concurrent access

Signed-off-by: PengFei Li <[email protected]>
banmoy · Aug 21, 2023 · 3314894 · 3314894
1 parent b3ae21e
commit 3314894
Showing 1 changed file with 9 additions and 14 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
@@ -35,10 +35,10 @@
 
 import java.io.Serializable;
 import java.nio.charset.StandardCharsets;
-import java.util.HashMap;
-import java.util.LinkedList;
 import java.util.Map;
 import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
@@ -84,7 +84,7 @@ enum State {
     private final long maxCacheBytes;
     // threshold to block write
     private final long maxWriteBlockCacheBytes;
-    private final Map<String, TableRegion> regions = new HashMap<>();
+    private final Map<String, TableRegion> regions = new ConcurrentHashMap<>();
     private final AtomicLong currentCacheBytes = new AtomicLong(0L);
     private final AtomicLong totalFlushRows = new AtomicLong(0L);
 
@@ -105,7 +105,7 @@ enum State {
     private final AtomicReference<State> state = new AtomicReference<>(State.INACTIVE);
     private volatile Throwable e;
 
-    private final Queue<TransactionTableRegion> flushQ = new LinkedList<>();
+    private final Queue<TransactionTableRegion> flushQ = new ConcurrentLinkedQueue<>();
 
     /**
      * Whether write() has triggered a flush after currentCacheBytes > maxCacheBytes.
@@ -419,16 +419,11 @@ protected TableRegion getCacheRegion(String uniqueKey, String database, String t
 
         TableRegion region = regions.get(uniqueKey);
         if (region == null) {
-            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);
-                }
-            }
+            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;
     }