[Hotfix][Zeta][Checkpoint] Fix CheckpointIDCounter Thread Not Safe Pr…

…oblem (#3875) * [Improve] [Checkpoint] Fix CheckpointIDCounter Thread Not Safe Problem * [Improve] [Checkpoint] Fix CheckpointIDCounter Thread Not Safe Problem
apache · Jan 6, 2023 · 41751d8 · 41751d8
1 parent 73fc0c1
commit 41751d8
Showing 1 changed file with 5 additions and 3 deletions.
diff --git a/.../src/main/java/org/apache/seatunnel/engine/server/checkpoint/IMapCheckpointIDCounter.java b/.../src/main/java/org/apache/seatunnel/engine/server/checkpoint/IMapCheckpointIDCounter.java
@@ -17,6 +17,8 @@
 
 package org.apache.seatunnel.engine.server.checkpoint;
 
+import static com.google.common.base.Preconditions.checkNotNull;
+
 import org.apache.seatunnel.common.utils.RetryUtils;
 import org.apache.seatunnel.engine.common.Constant;
 import org.apache.seatunnel.engine.core.checkpoint.CheckpointIDCounter;
@@ -55,9 +57,9 @@ public CompletableFuture<Void> shutdown(PipelineStatus pipelineStatus) {
 
     @Override
     public long getAndIncrement() throws Exception {
-        Long currentId = checkpointIdMap.get(pipelineId);
-        checkpointIdMap.put(pipelineId, currentId + 1);
-        return currentId;
+        Long nextId = checkpointIdMap.compute(pipelineId, (k, v) -> v == null ? null : v + 1);
+        checkNotNull(nextId);
+        return nextId - 1;
     }
 
     @Override