[Hotfix][Connector-V2] Fix ConcurrentModificationException when snaps…

…hotState based on SourceReaderBase (#4011)
apache · Feb 1, 2023 · cd2bd6a · cd2bd6a
1 parent 79b5cdd
commit cd2bd6a
Showing 1 changed file with 4 additions and 2 deletions.
diff --git a/...java/org/apache/seatunnel/connectors/seatunnel/common/source/reader/SourceReaderBase.java b/...java/org/apache/seatunnel/connectors/seatunnel/common/source/reader/SourceReaderBase.java
@@ -36,6 +36,8 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 
 /**
  * An abstract implementation of {@link SourceReader} which provides some synchronization between
@@ -55,7 +57,7 @@
 public abstract class SourceReaderBase<E, T, SplitT extends SourceSplit, SplitStateT>
     implements SourceReader<T, SplitT> {
     private final BlockingQueue<RecordsWithSplitIds<E>> elementsQueue;
-    private final Map<String, SplitContext<T, SplitStateT>> splitStates;
+    private final ConcurrentMap<String, SplitContext<T, SplitStateT>> splitStates;
     protected final RecordEmitter<E, T, SplitStateT> recordEmitter;
     protected final SplitFetcherManager<E, SplitT> splitFetcherManager;
     protected final SourceReaderOptions options;
@@ -74,7 +76,7 @@ public SourceReaderBase(BlockingQueue<RecordsWithSplitIds<E>> elementsQueue,
         this.elementsQueue = elementsQueue;
         this.splitFetcherManager = splitFetcherManager;
         this.recordEmitter = recordEmitter;
-        this.splitStates = new HashMap<>();
+        this.splitStates = new ConcurrentHashMap<>();
         this.options = options;
         this.context = context;
     }