[Bugfix][CDC Base] Solving the ConcurrentModificationException caused…

… by snapshotState being modified concurrently. (#4877)
apache · Jun 5, 2023 · 9a2efa5 · 9a2efa5
1 parent f4077cb
commit 9a2efa5
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 4 deletions.
diff --git a/...g/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSourceEnumerator.java b/...g/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSourceEnumerator.java
@@ -109,7 +109,9 @@ public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) {
                     (CompletedSnapshotSplitsReportEvent) sourceEvent;
             List<SnapshotSplitWatermark> completedSplitWatermarks =
                     reportEvent.getCompletedSnapshotSplitWatermarks();
-            splitAssigner.onCompletedSplits(completedSplitWatermarks);
+            synchronized (context) {
+                splitAssigner.onCompletedSplits(completedSplitWatermarks);
+            }
 
             // send acknowledge event
             CompletedSnapshotSplitsAckEvent ackEvent =
@@ -153,7 +155,10 @@ private void assignSplits() {
                 continue;
             }
 
-            Optional<SourceSplitBase> split = splitAssigner.getNext();
+            Optional<SourceSplitBase> split;
+            synchronized (context) {
+                split = splitAssigner.getNext();
+            }
             if (split.isPresent()) {
                 final SourceSplitBase sourceSplit = split.get();
                 context.assignSplit(nextAwaiting, sourceSplit);

diff --git a/...g/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/source/offset/LsnOffset.java b/...g/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/source/offset/LsnOffset.java
@@ -61,8 +61,8 @@ public Lsn getCommitLsn() {
         return Lsn.valueOf(offset.get(SourceInfo.COMMIT_LSN_KEY));
     }
 
-    public Long getEventSerialNo() {
-        return Long.valueOf(offset.get(SourceInfo.EVENT_SERIAL_NO_KEY));
+    public Object getEventSerialNo() {
+        return offset.get(SourceInfo.EVENT_SERIAL_NO_KEY);
     }
 
     public int compareTo(Offset o) {

diff --git a/...rver/src/main/java/org/apache/seatunnel/engine/server/task/SourceSplitEnumeratorTask.java b/...rver/src/main/java/org/apache/seatunnel/engine/server/task/SourceSplitEnumeratorTask.java
@@ -142,6 +142,7 @@ public void triggerBarrier(Barrier barrier) throws Exception {
         final long barrierId = barrier.getId();
         Serializable snapshotState = null;
         byte[] serialize = null;
+        // Do not modify this lock object, as it is also used in the SourceSplitEnumerator.
         synchronized (enumeratorContext) {
             if (barrier.snapshot()) {
                 snapshotState = enumerator.snapshotState(barrierId);