[incubator-seatunnel] branch st-engine updated: check slot active before deploy task (#2741)

[fanjia]

This is an automated email from the ASF dual-hosted git repository.

fanjia pushed a commit to branch st-engine
in repository https://gitbox.apache.org/repos/asf/incubator-seatunnel.git


The following commit(s) were added to refs/heads/st-engine by this push:
     new ac1a73467 check slot active before deploy task (#2741)
ac1a73467 is described below

commit ac1a734672ca1235ba6b4bb98888db897db4bbee
Author: Eric <gaojun2...@gmail.com>
AuthorDate: Fri Sep 16 10:14:57 2022 +0800

    check slot active before deploy task (#2741)
---
 .../seatunnel/engine/server/scheduler/PipelineBaseScheduler.java    | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git 
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/scheduler/PipelineBaseScheduler.java
 
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/scheduler/PipelineBaseScheduler.java
index 52dd0410e..ebf0166eb 100644
--- 
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/scheduler/PipelineBaseScheduler.java
+++ 
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/scheduler/PipelineBaseScheduler.java
@@ -125,8 +125,10 @@ public class PipelineBaseScheduler implements JobScheduler 
{
     private SlotProfile getOrApplyResourceForTask(@NonNull PhysicalVertex task,
                                                   Map<TaskGroupLocation, 
SlotProfile> ownedSlotProfiles) {
 
-        if (ownedSlotProfiles == null || ownedSlotProfiles.isEmpty() ||
-            ownedSlotProfiles.get(task.getTaskGroupLocation()) == null) {
+        if (ownedSlotProfiles == null
+            || ownedSlotProfiles.isEmpty()
+            || ownedSlotProfiles.get(task.getTaskGroupLocation()) == null
+            || 
!resourceManager.slotActiveCheck(ownedSlotProfiles.get(task.getTaskGroupLocation())))
 {
             return applyResourceForTask(task).join();
         }
         task.updateTaskState(ExecutionState.CREATED, ExecutionState.SCHEDULED);