Merge branch 'dev' into feature-seatunnel-simplify

dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/registry/AlertHeartbeatTask.java

@@ -75,6 +75,7 @@ public AlertServerHeartBeat getHeartBeat() {
                 .cpuUsage(systemMetrics.getSystemCpuUsagePercentage())
                 .memoryUsage(systemMetrics.getSystemMemoryUsedPercentage())
                 .jvmMemoryUsage(systemMetrics.getJvmMemoryUsedPercentage())
+                .diskUsage(systemMetrics.getDiskUsedPercentage())
                 .serverStatus(ServerStatus.NORMAL)
                 .isActive(alertHAServer.isActive())
                 .host(NetUtils.getHost())

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/MonitorController.java

@@ -65,7 +65,7 @@ public class MonitorController extends BaseController {
     @ResponseStatus(HttpStatus.OK)
     @ApiException(LIST_MASTERS_ERROR)
     public Result<List<Server>> listServer(@PathVariable("nodeType") RegistryNodeType nodeType) {
-        List<Server> servers = monitorService.listServer(nodeType);
+        final List<Server> servers = monitorService.listServer(nodeType);
         return Result.success(servers);
     }
 

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java

@@ -33,9 +33,11 @@
 import org.apache.dolphinscheduler.api.service.UsersService;
 import org.apache.dolphinscheduler.api.service.WorkflowDefinitionService;
 import org.apache.dolphinscheduler.common.constants.Constants;
+import org.apache.dolphinscheduler.common.enums.CommandType;
 import org.apache.dolphinscheduler.common.enums.ComplementDependentMode;
 import org.apache.dolphinscheduler.common.enums.ExecutionOrder;
 import org.apache.dolphinscheduler.common.enums.FailureStrategy;
+import org.apache.dolphinscheduler.common.enums.Flag;
 import org.apache.dolphinscheduler.common.enums.Priority;
 import org.apache.dolphinscheduler.common.enums.ReleaseState;
 import org.apache.dolphinscheduler.common.enums.RunMode;
@@ -370,11 +372,9 @@ private void createOrUpdateSchedule(User user,
     public void execWorkflowInstance(String userName,
                                      String projectName,
                                      String workflowName,
-                                     String cronTime,
                                      String workerGroup,
                                      String warningType,
-                                     Integer warningGroupId,
-                                     Integer timeout) {
+                                     Integer warningGroupId) {
         User user = usersService.queryUser(userName);
         Project project = projectMapper.queryByName(projectName);
         WorkflowDefinition workflowDefinition =
@@ -389,6 +389,10 @@ public void execWorkflowInstance(String userName,
                 .workerGroup(workerGroup)
                 .warningType(WarningType.of(warningType))
                 .warningGroupId(warningGroupId)
+                .execType(CommandType.START_PROCESS)
+                .taskDependType(TaskDependType.TASK_POST)
+                .dryRun(Flag.NO)
+                .testFlag(Flag.NO)
                 .build();
         executorService.triggerWorkflowDefinition(workflowTriggerRequest);
     }

dolphinscheduler-eventbus/src/main/java/org/apache/dolphinscheduler/eventbus/AbstractDelayEvent.java

@@ -39,6 +39,10 @@ public abstract class AbstractDelayEvent implements IEvent, Delayed {
     @Builder.Default
     protected long createTimeInNano = System.nanoTime();
 
+    // set create time as default if the inheritor didn't call super()
+    @Builder.Default
+    protected long expiredTimeInNano = System.nanoTime();
+
     public AbstractDelayEvent() {
         this(DEFAULT_DELAY_TIME);
     }
@@ -50,6 +54,7 @@ public AbstractDelayEvent(final long delayTime) {
     public AbstractDelayEvent(final long delayTime, final long createTimeInNano) {
         this.delayTime = delayTime;
         this.createTimeInNano = createTimeInNano;
+        this.expiredTimeInNano = this.delayTime * 1_000_000 + this.createTimeInNano;
     }
 
     @Override
@@ -60,7 +65,7 @@ public long getDelay(TimeUnit unit) {
 
     @Override
     public int compareTo(Delayed other) {
-        return Long.compare(this.createTimeInNano, ((AbstractDelayEvent) other).createTimeInNano);
+        return Long.compare(this.expiredTimeInNano, ((AbstractDelayEvent) other).expiredTimeInNano);
     }
 
 }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/cluster/ClusterManager.java

@@ -17,6 +17,9 @@
 
 package org.apache.dolphinscheduler.server.master.cluster;
 
+import org.apache.dolphinscheduler.common.model.MasterHeartBeat;
+import org.apache.dolphinscheduler.common.model.WorkerHeartBeat;
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
 import org.apache.dolphinscheduler.registry.api.RegistryClient;
 import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
 
@@ -36,6 +39,9 @@ public class ClusterManager {
     @Getter
     private WorkerClusters workerClusters;
 
+    @Autowired
+    private MasterSlotManager masterSlotManager;
+
     @Autowired
     private WorkerGroupChangeNotifier workerGroupChangeNotifier;
 
@@ -48,11 +54,48 @@ public ClusterManager() {
     }
 
     public void start() {
+        initializeMasterClusters();
+        initializeWorkerClusters();
+        log.info("ClusterManager started...");
+    }
+
+    /**
+     * Initialize the master clusters.
+     * <p> 1. Register master slot listener once master clusters changed.
+     * <p> 2. Fetch master nodes from registry.
+     * <p> 3. Subscribe the master change event.
+     */
+    private void initializeMasterClusters() {
+        this.masterClusters.registerListener(new MasterSlotChangeListenerAdaptor(masterSlotManager, masterClusters));
+
+        registryClient.getServerList(RegistryNodeType.MASTER).forEach(server -> {
+            final MasterHeartBeat masterHeartBeat =
+                    JSONUtils.parseObject(server.getHeartBeatInfo(), MasterHeartBeat.class);
+            masterClusters.onServerAdded(MasterServerMetadata.parseFromHeartBeat(masterHeartBeat));
+        });
+        log.info("Initialized MasterClusters: {}", JSONUtils.toPrettyJsonString(masterClusters.getServers()));
+
         this.registryClient.subscribe(RegistryNodeType.MASTER.getRegistryPath(), masterClusters);
+    }
+
+    /**
+     * Initialize the worker clusters.
+     * <p> 1. Fetch worker nodes from registry.
+     * <p> 2. Register worker group change notifier once worker clusters changed.
+     * <p> 3. Subscribe the worker change event.
+     */
+    private void initializeWorkerClusters() {
+        registryClient.getServerList(RegistryNodeType.WORKER).forEach(server -> {
+            final WorkerHeartBeat workerHeartBeat =
+                    JSONUtils.parseObject(server.getHeartBeatInfo(), WorkerHeartBeat.class);
+            workerClusters.onServerAdded(WorkerServerMetadata.parseFromHeartBeat(workerHeartBeat));
+        });
+        log.info("Initialized WorkerClusters: {}", JSONUtils.toPrettyJsonString(workerClusters.getServers()));
+
         this.registryClient.subscribe(RegistryNodeType.WORKER.getRegistryPath(), workerClusters);
+
         this.workerGroupChangeNotifier.subscribeWorkerGroupsChange(workerClusters);
         this.workerGroupChangeNotifier.start();
-        log.info("ClusterManager started...");
     }
 
 }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/cluster/IMasterSlotChangeListener.java

@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.server.master.cluster;
+
+import java.util.List;
+
+public interface IMasterSlotChangeListener {
+
+    void onMasterSlotChanged(final List<MasterServerMetadata> normalMasterServers);
+}

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/cluster/MasterServerMetadata.java

@@ -17,6 +17,8 @@
 
 package org.apache.dolphinscheduler.server.master.cluster;
 
+import static com.google.common.base.Preconditions.checkNotNull;
+
 import org.apache.dolphinscheduler.common.constants.Constants;
 import org.apache.dolphinscheduler.common.model.MasterHeartBeat;
 
@@ -32,6 +34,7 @@
 public class MasterServerMetadata extends BaseServerMetadata implements Comparable<MasterServerMetadata> {
 
     public static MasterServerMetadata parseFromHeartBeat(final MasterHeartBeat masterHeartBeat) {
+        checkNotNull(masterHeartBeat);
         return MasterServerMetadata.builder()
                 .processId(masterHeartBeat.getProcessId())
                 .serverStartupTime(masterHeartBeat.getStartupTime())

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/cluster/MasterSlotChangeListenerAdaptor.java

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.server.master.cluster;
+
+import java.util.List;
+
+public class MasterSlotChangeListenerAdaptor
+        implements
+            IMasterSlotChangeListener,
+            IClusters.IClustersChangeListener<MasterServerMetadata> {
+
+    private final MasterSlotManager masterSlotManager;
+
+    private final MasterClusters masterClusters;
+
+    public MasterSlotChangeListenerAdaptor(final MasterSlotManager masterSlotManager,
+                                           final MasterClusters masterClusters) {
+        this.masterSlotManager = masterSlotManager;
+        this.masterClusters = masterClusters;
+    }
+
+    @Override
+    public void onMasterSlotChanged(final List<MasterServerMetadata> normalMasterServers) {
+        masterSlotManager.doReBalance(normalMasterServers);
+    }
+
+    @Override
+    public void onServerAdded(MasterServerMetadata server) {
+        onMasterSlotChanged(masterClusters.getNormalServers());
+    }
+
+    @Override
+    public void onServerRemove(MasterServerMetadata server) {
+        onMasterSlotChanged(masterClusters.getNormalServers());
+    }
+
+    @Override
+    public void onServerUpdate(MasterServerMetadata server) {
+        onMasterSlotChanged(masterClusters.getNormalServers());
+    }
+}

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/cluster/MasterSlotManager.java

@@ -29,34 +29,14 @@
 @Component
 public class MasterSlotManager implements IMasterSlotReBalancer {
 
-    private final MasterClusters masterClusters;
-
     private final MasterConfig masterConfig;
 
     private volatile int currentSlot = -1;
 
     private volatile int totalSlots = 0;
 
-    public MasterSlotManager(ClusterManager clusterManager, MasterConfig masterConfig) {
+    public MasterSlotManager(final MasterConfig masterConfig) {
         this.masterConfig = masterConfig;
-        this.masterClusters = clusterManager.getMasterClusters();
-        this.masterClusters.registerListener(new IClusters.IClustersChangeListener<MasterServerMetadata>() {
-
-            @Override
-            public void onServerAdded(MasterServerMetadata server) {
-                doReBalance(masterClusters.getNormalServers());
-            }
-
-            @Override
-            public void onServerRemove(MasterServerMetadata server) {
-                doReBalance(masterClusters.getNormalServers());
-            }
-
-            @Override
-            public void onServerUpdate(MasterServerMetadata server) {
-                doReBalance(masterClusters.getNormalServers());
-            }
-        });
     }
 
     /**

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/IWorkflowExecutionGraph.java

@@ -191,6 +191,11 @@ public interface IWorkflowExecutionGraph {
      */
     boolean isTaskExecutionRunnableForbidden(final ITaskExecutionRunnable taskExecutionRunnable);
 
+    /**
+     * Whether the given task's execution is failure and waiting for retry.
+     */
+    boolean isTaskExecutionRunnableRetrying(final ITaskExecutionRunnable taskExecutionRunnable);
+
     /**
      * Whether all predecessors task is skipped.
      * <p> Once all predecessors are marked as skipped, then the task will be marked as skipped, and will trigger its successors.

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowExecutionGraph.java

@@ -140,7 +140,7 @@ public ITaskExecutionRunnable getTaskExecutionRunnableByTaskCode(final Long task
 
     @Override
     public boolean isTaskExecutionRunnableActive(final ITaskExecutionRunnable taskExecutionRunnable) {
-        return activeTaskExecutionRunnable.add(taskExecutionRunnable.getName());
+        return activeTaskExecutionRunnable.contains(taskExecutionRunnable.getName());
     }
 
     @Override
@@ -256,6 +256,16 @@ public boolean isTaskExecutionRunnableForbidden(final ITaskExecutionRunnable tas
         return (taskExecutionRunnable.getTaskDefinition().getFlag() == Flag.NO);
     }
 
+    @Override
+    public boolean isTaskExecutionRunnableRetrying(final ITaskExecutionRunnable taskExecutionRunnable) {
+        if (!taskExecutionRunnable.isTaskInstanceInitialized()) {
+            return false;
+        }
+        final TaskInstance taskInstance = taskExecutionRunnable.getTaskInstance();
+        return taskInstance.getState() == TaskExecutionStatus.FAILURE && taskExecutionRunnable.isTaskInstanceCanRetry()
+                && isTaskExecutionRunnableActive(taskExecutionRunnable);
+    }
+
     /**
      * Whether all predecessors are skipped.
      * <p> Only when all predecessors are skipped, will return true. If the given task doesn't have any predecessors, will return false.

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskFailureStateAction.java

@@ -104,6 +104,12 @@ public void pauseEventAction(final IWorkflowExecutionRunnable workflowExecutionR
                                  final ITaskExecutionRunnable taskExecutionRunnable,
                                  final TaskPauseLifecycleEvent taskPauseEvent) {
         throwExceptionIfStateIsNotMatch(taskExecutionRunnable);
+        // When the failed task is awaiting retry, we can mark it as 'paused' to ignore the retry event.
+        if (isTaskRetrying(taskExecutionRunnable)) {
+            super.pausedEventAction(workflowExecutionRunnable, taskExecutionRunnable,
+                    TaskPausedLifecycleEvent.of(taskExecutionRunnable));
+            return;
+        }
         logWarningIfCannotDoAction(taskExecutionRunnable, taskPauseEvent);
     }
 
@@ -112,14 +118,11 @@ public void pausedEventAction(final IWorkflowExecutionRunnable workflowExecution
                                   final ITaskExecutionRunnable taskExecutionRunnable,
                                   final TaskPausedLifecycleEvent taskPausedEvent) {
         throwExceptionIfStateIsNotMatch(taskExecutionRunnable);
-        final IWorkflowExecutionGraph workflowExecutionGraph = taskExecutionRunnable.getWorkflowExecutionGraph();
         // This case happen when the task is failure but the task is in delay retry queue.
         // We don't remove the event in GlobalWorkflowDelayEventCoordinator the event should be dropped when the task is
         // killed.
-        if (taskExecutionRunnable.isTaskInstanceCanRetry()
-                && workflowExecutionGraph.isTaskExecutionRunnableActive(taskExecutionRunnable)) {
-            workflowExecutionGraph.markTaskExecutionRunnableChainPause(taskExecutionRunnable);
-            publishWorkflowInstanceTopologyLogicalTransitionEvent(taskExecutionRunnable);
+        if (isTaskRetrying(taskExecutionRunnable)) {
+            super.pausedEventAction(workflowExecutionRunnable, taskExecutionRunnable, taskPausedEvent);
             return;
         }
         logWarningIfCannotDoAction(taskExecutionRunnable, taskPausedEvent);
@@ -130,6 +133,12 @@ public void killEventAction(final IWorkflowExecutionRunnable workflowExecutionRu
                                 final ITaskExecutionRunnable taskExecutionRunnable,
                                 final TaskKillLifecycleEvent taskKillEvent) {
         throwExceptionIfStateIsNotMatch(taskExecutionRunnable);
+        // When the failed task is awaiting retry, we can mark it as 'killed' to ignore the retry event.
+        if (isTaskRetrying(taskExecutionRunnable)) {
+            super.killedEventAction(workflowExecutionRunnable, taskExecutionRunnable,
+                    TaskKilledLifecycleEvent.of(taskExecutionRunnable));
+            return;
+        }
         logWarningIfCannotDoAction(taskExecutionRunnable, taskKillEvent);
     }
 
@@ -138,14 +147,11 @@ public void killedEventAction(final IWorkflowExecutionRunnable workflowExecution
                                   final ITaskExecutionRunnable taskExecutionRunnable,
                                   final TaskKilledLifecycleEvent taskKilledEvent) {
         throwExceptionIfStateIsNotMatch(taskExecutionRunnable);
-        final IWorkflowExecutionGraph workflowExecutionGraph = taskExecutionRunnable.getWorkflowExecutionGraph();
         // This case happen when the task is failure but the task is in delay retry queue.
         // We don't remove the event in GlobalWorkflowDelayEventCoordinator the event should be dropped when the task is
         // killed.
-        if (taskExecutionRunnable.isTaskInstanceCanRetry()
-                && workflowExecutionGraph.isTaskExecutionRunnableActive(taskExecutionRunnable)) {
-            workflowExecutionGraph.markTaskExecutionRunnableChainKill(taskExecutionRunnable);
-            publishWorkflowInstanceTopologyLogicalTransitionEvent(taskExecutionRunnable);
+        if (isTaskRetrying(taskExecutionRunnable)) {
+            super.killedEventAction(workflowExecutionRunnable, taskExecutionRunnable, taskKilledEvent);
             return;
         }
         logWarningIfCannotDoAction(taskExecutionRunnable, taskKilledEvent);
@@ -179,4 +185,9 @@ public void failoverEventAction(final IWorkflowExecutionRunnable workflowExecuti
     public TaskExecutionStatus matchState() {
         return TaskExecutionStatus.FAILURE;
     }
+
+    private boolean isTaskRetrying(final ITaskExecutionRunnable taskExecutionRunnable) {
+        final IWorkflowExecutionGraph workflowExecutionGraph = taskExecutionRunnable.getWorkflowExecutionGraph();
+        return workflowExecutionGraph.isTaskExecutionRunnableRetrying(taskExecutionRunnable);
+    }
 }