Add logs to find job restore from master active switch error

apache · Jul 15, 2023 · c49a449 · c49a449
1 parent a16cd57
commit c49a449
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 4 deletions.
diff --git a/...server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointManager.java b/...server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointManager.java
@@ -47,6 +47,7 @@
 import com.hazelcast.spi.impl.operationservice.impl.InvocationFuture;
 import lombok.extern.slf4j.Slf4j;
 
+import java.util.Arrays;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
@@ -167,6 +168,9 @@ public PassiveCompletableFuture<CompletedCheckpoint> triggerSavepoint(int pipeli
     }
 
     public void reportedPipelineRunning(int pipelineId, boolean alreadyStarted) {
+        log.info(
+                "reported pipeline running stack: "
+                        + Arrays.toString(Thread.currentThread().getStackTrace()));
         getCheckpointCoordinator(pipelineId).restoreCoordinator(alreadyStarted);
     }
 

diff --git a/...-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java b/...-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java
@@ -394,7 +394,7 @@ private boolean turnToEndState(@NonNull ExecutionState endState) {
     public boolean updateTaskState(
             @NonNull ExecutionState current, @NonNull ExecutionState targetState) {
         synchronized (this) {
-            LOGGER.fine(
+            LOGGER.info(
                     String.format(
                             "Try to update the task %s state from %s to %s",
                             taskFullName, current, targetState));

diff --git a/...-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/SubPlan.java b/...-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/SubPlan.java
@@ -53,7 +53,7 @@ public class SubPlan {
     private static final ILogger LOGGER = Logger.getLogger(SubPlan.class);
 
     /** The max num pipeline can restore. */
-    public static final int PIPELINE_MAX_RESTORE_NUM = 2; // TODO should set by config
+    public static final int PIPELINE_MAX_RESTORE_NUM = 3; // TODO should set by config
 
     private final List<PhysicalVertex> physicalVertexList;
 
@@ -332,6 +332,9 @@ private void turnToEndState(@NonNull PipelineStatus endState) throws Exception {
                             exception -> ExceptionUtil.isOperationNeedRetryException(exception),
                             Constant.OPERATION_RETRY_SLEEP));
             this.currPipelineStatus = endState;
+            LOGGER.info(
+                    String.format(
+                            "%s turn to end state %s.", pipelineFullName, currPipelineStatus));
         }
     }
 
@@ -511,11 +514,17 @@ private void resetPipelineState() throws Exception {
                         LOGGER.severe(message);
                         throw new IllegalStateException(message);
                     }
-
+                    LOGGER.info(
+                            String.format(
+                                    "Reset pipeline %s state to %s",
+                                    getPipelineFullName(), PipelineStatus.CREATED));
                     updateStateTimestamps(PipelineStatus.CREATED);
                     runningJobStateIMap.set(pipelineLocation, PipelineStatus.CREATED);
                     this.currPipelineStatus = PipelineStatus.CREATED;
-                    ;
+                    LOGGER.info(
+                            String.format(
+                                    "Reset pipeline %s state to %s complete",
+                                    getPipelineFullName(), PipelineStatus.CREATED));
                     return null;
                 },
                 new RetryUtils.RetryMaterial(