You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by zi...@apache.org on 2026/07/17 06:58:59 UTC
(dolphinscheduler) branch dev updated: [Fix-18338] Check task if it's waiting for TaskGroup slot when pause/kill (#18414)
This is an automated email from the ASF dual-hosted git repository.
SbloodyS pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new c1cad902a9 [Fix-18338] Check task if it's waiting for TaskGroup slot when pause/kill (#18414)
c1cad902a9 is described below
commit c1cad902a986872fd022da6ba3c4628626241cd0
Author: xiangzihao <46...@qq.com>
AuthorDate: Fri Jul 17 14:58:53 2026 +0800
[Fix-18338] Check task if it's waiting for TaskGroup slot when pause/kill (#18414)
---
.../dao/mapper/TaskGroupQueueMapper.java | 11 ++
.../dao/repository/TaskGroupQueueDao.java | 20 ++++
.../dao/repository/impl/TaskGroupQueueDaoImpl.java | 17 +++
.../dao/mapper/TaskGroupQueueMapper.xml | 15 +++
.../master/engine/ITaskGroupCoordinator.java | 11 ++
.../server/master/engine/TaskGroupCoordinator.java | 117 +++++++++++++++----
.../statemachine/TaskSubmittedStateAction.java | 12 ++
.../cases/WorkflowStartTaskGroupTestCase.java | 129 +++++++++++++++++++++
8 files changed, 308 insertions(+), 24 deletions(-)
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskGroupQueueMapper.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskGroupQueueMapper.java
index c4e6495e90..b9f16534cf 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskGroupQueueMapper.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskGroupQueueMapper.java
@@ -22,6 +22,7 @@ import org.apache.dolphinscheduler.dao.entity.TaskGroupQueue;
import org.apache.ibatis.annotations.Param;
+import java.util.Date;
import java.util.List;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
@@ -118,6 +119,16 @@ public interface TaskGroupQueueMapper extends BaseMapper<TaskGroupQueue> {
List<TaskGroupQueue> queryByTaskInstanceId(@Param("taskInstanceId") Integer taskInstanceId);
+ int acquireTaskGroupQueue(@Param("id") Integer id,
+ @Param("waitingStatus") int waitingStatus,
+ @Param("acquiredStatus") int acquiredStatus,
+ @Param("inQueue") int inQueue,
+ @Param("forceStart") int forceStart,
+ @Param("updateTime") Date updateTime);
+
+ int deleteByTaskInstanceIdAndStatus(@Param("taskInstanceId") Integer taskInstanceId,
+ @Param("status") int status);
+
List<TaskGroupQueue> queryUsingTaskGroupQueueByGroupId(@Param("taskGroupId") Integer taskGroupId,
@Param("status") int status,
@Param("inQueue") int inQueue,
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/TaskGroupQueueDao.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/TaskGroupQueueDao.java
index 809a6a4f17..78570eb8f5 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/TaskGroupQueueDao.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/TaskGroupQueueDao.java
@@ -17,9 +17,11 @@
package org.apache.dolphinscheduler.dao.repository;
+import org.apache.dolphinscheduler.common.enums.TaskGroupQueueStatus;
import org.apache.dolphinscheduler.dao.entity.TaskGroupQueue;
import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
+import java.util.Date;
import java.util.List;
public interface TaskGroupQueueDao extends IDao<TaskGroupQueue> {
@@ -65,6 +67,24 @@ public interface TaskGroupQueueDao extends IDao<TaskGroupQueue> {
*/
List<TaskGroupQueue> queryByTaskInstanceId(Integer taskInstanceId);
+ /**
+ * Atomically change a waiting TaskGroupQueue to acquired.
+ *
+ * @param id taskGroupQueue id
+ * @param updateTime update time
+ * @return true if the waiting TaskGroupQueue was acquired
+ */
+ boolean acquireTaskGroupQueue(Integer id, Date updateTime);
+
+ /**
+ * Delete the TaskGroupQueue only when its current status matches the expected status.
+ *
+ * @param taskInstanceId taskInstance id
+ * @param status expected status
+ * @return true if a TaskGroupQueue was deleted
+ */
+ boolean deleteByTaskInstanceIdAndStatus(Integer taskInstanceId, TaskGroupQueueStatus status);
+
/**
* Query all {@link TaskGroupQueue} which status is TaskGroupQueueStatus.ACQUIRE_SUCCESS and forceStart is {@link org.apache.dolphinscheduler.common.enums.Flag#NO}.
*
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/TaskGroupQueueDaoImpl.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/TaskGroupQueueDaoImpl.java
index 5fd50deaae..979fae432e 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/TaskGroupQueueDaoImpl.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/TaskGroupQueueDaoImpl.java
@@ -26,6 +26,7 @@ import org.apache.dolphinscheduler.dao.repository.TaskGroupQueueDao;
import org.apache.commons.collections4.CollectionUtils;
+import java.util.Date;
import java.util.List;
import lombok.NonNull;
@@ -67,6 +68,22 @@ public class TaskGroupQueueDaoImpl extends BaseDao<TaskGroupQueue, TaskGroupQueu
return mybatisMapper.queryByTaskInstanceId(taskInstanceId);
}
+ @Override
+ public boolean acquireTaskGroupQueue(Integer id, Date updateTime) {
+ return mybatisMapper.acquireTaskGroupQueue(
+ id,
+ TaskGroupQueueStatus.WAIT_QUEUE.getCode(),
+ TaskGroupQueueStatus.ACQUIRE_SUCCESS.getCode(),
+ Flag.YES.getCode(),
+ Flag.NO.getCode(),
+ updateTime) > 0;
+ }
+
+ @Override
+ public boolean deleteByTaskInstanceIdAndStatus(Integer taskInstanceId, TaskGroupQueueStatus status) {
+ return mybatisMapper.deleteByTaskInstanceIdAndStatus(taskInstanceId, status.getCode()) > 0;
+ }
+
@Override
public List<TaskGroupQueue> queryAcquiredTaskGroupQueueByGroupId(Integer taskGroupId) {
return mybatisMapper.queryUsingTaskGroupQueueByGroupId(
diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskGroupQueueMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskGroupQueueMapper.xml
index 71d60b8a2c..fc5518df8f 100644
--- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskGroupQueueMapper.xml
+++ b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskGroupQueueMapper.xml
@@ -236,6 +236,21 @@
where task_id = #{taskInstanceId}
</select>
+ <update id="acquireTaskGroupQueue">
+ update t_ds_task_group_queue
+ set status = #{acquiredStatus}, update_time = #{updateTime}
+ where id = #{id}
+ and status = #{waitingStatus}
+ and in_queue = #{inQueue}
+ and force_start = #{forceStart}
+ </update>
+
+ <delete id="deleteByTaskInstanceIdAndStatus">
+ delete from t_ds_task_group_queue
+ where task_id = #{taskInstanceId}
+ and status = #{status}
+ </delete>
+
<select id="queryUsingTaskGroupQueueByGroupId" resultType="org.apache.dolphinscheduler.dao.entity.TaskGroupQueue">
select
<include refid="baseSql" />
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/ITaskGroupCoordinator.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/ITaskGroupCoordinator.java
index 6fd885a21f..854097b021 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/ITaskGroupCoordinator.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/ITaskGroupCoordinator.java
@@ -91,6 +91,17 @@ public interface ITaskGroupCoordinator extends AutoCloseable {
*/
void releaseTaskGroupSlot(TaskInstance taskInstance);
+ /**
+ * Remove the task from the TaskGroup waiting queue.
+ * <p>
+ * The removal succeeds only while the queue is still in {@link TaskGroupQueueStatus#WAIT_QUEUE}. A successful
+ * removal guarantees that the coordinator has not acquired the queue for dispatch.
+ *
+ * @param taskInstance taskInstance
+ * @return true if the task was removed from the waiting queue
+ */
+ boolean releaseWaitingTaskGroupSlot(TaskInstance taskInstance);
+
/**
* Close the TaskGroupCoordinator, once closed, the coordinator will not work until you have started the coordinator again.
*/
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/TaskGroupCoordinator.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/TaskGroupCoordinator.java
index 1a0312c70f..3a6d81c24c 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/TaskGroupCoordinator.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/TaskGroupCoordinator.java
@@ -54,6 +54,7 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
+import org.springframework.transaction.support.TransactionTemplate;
import com.google.common.annotations.VisibleForTesting;
@@ -73,6 +74,9 @@ public class TaskGroupCoordinator implements ITaskGroupCoordinator, AutoCloseabl
@Autowired
private WorkflowInstanceDao workflowInstanceDao;
+ @Autowired
+ private TransactionTemplate transactionTemplate;
+
private boolean flag = false;
private Thread internalThread;
@@ -233,16 +237,17 @@ public class TaskGroupCoordinator implements ITaskGroupCoordinator, AutoCloseabl
for (final TaskGroupQueue taskGroupQueue : taskGroupQueues) {
try {
LogUtils.setTaskInstanceIdMDC(taskGroupQueue.getTaskId());
- // notify the waiting task instance
- // We notify first, it notify failed, the taskGroupQueue will be in queue, and then we will retry it
- // next time.
- notifyWaitingTaskInstance(taskGroupQueue);
+ if (!notifyForceStartTaskGroupQueue(taskGroupQueue)) {
+ log.debug("Skip stale ForceStart TaskGroupQueue: {}", taskGroupQueue.getId());
+ continue;
+ }
log.info("Notify the ForceStart waiting TaskInstance: {} for taskGroupQueue: {} success",
taskGroupQueue.getTaskName(),
taskGroupQueue.getId());
-
- deleteTaskGroupQueueSlot(taskGroupQueue);
log.info("Release the force start TaskGroupQueue {}", taskGroupQueue);
+ } catch (RetryableTaskGroupNotificationException retryableException) {
+ log.info("Notify the ForceStart TaskGroupQueue: {} is temporarily unavailable, will retry",
+ taskGroupQueue.getId(), retryableException);
} catch (UnsupportedOperationException unsupportedOperationException) {
deleteTaskGroupQueueSlot(taskGroupQueue);
log.info(
@@ -256,6 +261,24 @@ public class TaskGroupCoordinator implements ITaskGroupCoordinator, AutoCloseabl
}
}
+ /**
+ * Delete the queue before notifying so only one coordinator can notify it. A notification failure rolls back the
+ * transaction and restores the queue for retry.
+ */
+ private boolean notifyForceStartTaskGroupQueue(TaskGroupQueue taskGroupQueue) {
+ final Boolean notified = transactionTemplate.execute(transactionStatus -> {
+ if (!taskGroupQueueDao.deleteById(taskGroupQueue.getId())) {
+ return false;
+ }
+ notifyWaitingTaskInstance(taskGroupQueue);
+ return true;
+ });
+ if (notified == null) {
+ throw new IllegalStateException("Notify ForceStart TaskGroupQueue transaction returned null");
+ }
+ return notified;
+ }
+
private void dealWithWaitingTaskGroupQueue() {
// Find the TaskGroup which usage < maxSize.
// Find the highest priority inQueue task group queue(Which is inQueue and status is Waiting and force start is
@@ -286,23 +309,15 @@ public class TaskGroupCoordinator implements ITaskGroupCoordinator, AutoCloseabl
for (TaskGroupQueue taskGroupQueue : taskGroupQueues) {
try {
LogUtils.setTaskInstanceIdMDC(taskGroupQueue.getTaskId());
- // Reduce the taskGroupSize
- boolean acquireResult = taskGroupDao.acquireTaskGroupSlot(taskGroup.getId());
- if (!acquireResult) {
- log.error("Failed to acquire task group slot for task group {}", taskGroup);
+ if (!acquireTaskGroupSlotAndNotify(taskGroupQueue)) {
+ log.debug("Skip stale or unavailable TaskGroupQueue: {} for TaskGroup: {}",
+ taskGroupQueue.getId(), taskGroup.getId());
continue;
}
- // Notify the waiting task instance
- // We notify first, it notify failed, the taskGroupQueue will be in queue, and then we will retry it
- // next time.
- notifyWaitingTaskInstance(taskGroupQueue);
-
- // Set the taskGroupQueue status to ACQUIRE_SUCCESS and remove from WAITING queue
- taskGroupQueue.setInQueue(Flag.YES.getCode());
- taskGroupQueue.setStatus(TaskGroupQueueStatus.ACQUIRE_SUCCESS);
- taskGroupQueue.setUpdateTime(new Date());
- taskGroupQueueDao.updateById(taskGroupQueue);
log.info("Success acquire TaskGroupSlot for TaskGroupQueue: {}", taskGroupQueue);
+ } catch (RetryableTaskGroupNotificationException retryableException) {
+ log.info("Notify Waiting TaskGroupQueue: {} is temporarily unavailable, will retry",
+ taskGroupQueue.getId(), retryableException);
} catch (UnsupportedOperationException unsupportedOperationException) {
deleteTaskGroupQueueSlot(taskGroupQueue);
log.info(
@@ -317,6 +332,24 @@ public class TaskGroupCoordinator implements ITaskGroupCoordinator, AutoCloseabl
}
}
+ private boolean acquireTaskGroupSlotAndNotify(TaskGroupQueue taskGroupQueue) {
+ final Boolean acquired = transactionTemplate.execute(transactionStatus -> {
+ if (!taskGroupDao.acquireTaskGroupSlot(taskGroupQueue.getGroupId())) {
+ return false;
+ }
+ if (!taskGroupQueueDao.acquireTaskGroupQueue(taskGroupQueue.getId(), new Date())) {
+ transactionStatus.setRollbackOnly();
+ return false;
+ }
+ notifyWaitingTaskInstance(taskGroupQueue);
+ return true;
+ });
+ if (acquired == null) {
+ throw new IllegalStateException("Acquire TaskGroupSlot transaction returned null");
+ }
+ return acquired;
+ }
+
@Override
public boolean needAcquireTaskGroupSlot(final TaskInstance taskInstance) {
if (taskInstance == null) {
@@ -393,6 +426,26 @@ public class TaskGroupCoordinator implements ITaskGroupCoordinator, AutoCloseabl
}
}
+ @Override
+ public boolean releaseWaitingTaskGroupSlot(TaskInstance taskInstance) {
+ if (taskInstance == null) {
+ throw new IllegalArgumentException("The TaskInstance is null");
+ }
+ if (taskInstance.getId() == null) {
+ throw new IllegalArgumentException("The TaskInstance id is null");
+ }
+ if (!TaskGroupUtils.isUsingTaskGroup(taskInstance)) {
+ return false;
+ }
+ final boolean removed = taskGroupQueueDao.deleteByTaskInstanceIdAndStatus(
+ taskInstance.getId(), TaskGroupQueueStatus.WAIT_QUEUE);
+ if (removed) {
+ log.info("Removed TaskInstance: {} from waiting TaskGroupQueue, taskGroupId: {}",
+ taskInstance.getId(), taskInstance.getTaskGroupId());
+ }
+ return removed;
+ }
+
private void notifyWaitingTaskInstance(TaskGroupQueue taskGroupQueue) {
// Find the related waiting task instance
// send RPC to notify the waiting task instance
@@ -414,12 +467,18 @@ public class TaskGroupCoordinator implements ITaskGroupCoordinator, AutoCloseabl
+ " is not exist, no need to notify");
}
if (workflowInstance.getState() != WorkflowExecutionStatus.RUNNING_EXECUTION) {
+ if (workflowInstance.getState() == WorkflowExecutionStatus.READY_PAUSE
+ || workflowInstance.getState() == WorkflowExecutionStatus.READY_STOP) {
+ throw new RetryableTaskGroupNotificationException(
+ "The WorkflowInstance: " + workflowInstance.getId() + " state is "
+ + workflowInstance.getState());
+ }
throw new UnsupportedOperationException(
"The WorkflowInstance: " + workflowInstance.getId() + " state is " + workflowInstance.getState()
+ ", no need to notify");
}
if (workflowInstance.getHost() == null || Constants.NULL.equals(workflowInstance.getHost())) {
- throw new UnsupportedOperationException(
+ throw new RetryableTaskGroupNotificationException(
"WorkflowInstance host is null, maybe it is in failover: " + workflowInstance);
}
@@ -435,16 +494,26 @@ public class TaskGroupCoordinator implements ITaskGroupCoordinator, AutoCloseabl
.withHost(workflowInstance.getHost())
.notifyTaskGroupSlotAcquireSuccess(taskGroupSlotAcquireSuccessNotifyRequest);
if (!taskGroupSlotAcquireSuccessNotifyResponse.isSuccess()) {
- throw new UnsupportedOperationException(
+ throw new RetryableTaskGroupNotificationException(
"Notify TaskInstance: " + taskInstance.getId() + " failed: "
+ taskGroupSlotAcquireSuccessNotifyResponse);
}
log.info("Wake up TaskInstance: {} success", taskInstance.getName());
}
+ private static final class RetryableTaskGroupNotificationException extends RuntimeException {
+
+ private RetryableTaskGroupNotificationException(String message) {
+ super(message);
+ }
+ }
+
private void deleteTaskGroupQueueSlot(TaskGroupQueue taskGroupQueue) {
- taskGroupQueueDao.deleteById(taskGroupQueue);
- log.info("Success release TaskGroupQueue: {}", taskGroupQueue);
+ if (taskGroupQueueDao.deleteById(taskGroupQueue)) {
+ log.info("Success release TaskGroupQueue: {}", taskGroupQueue);
+ return;
+ }
+ log.debug("TaskGroupQueue has already been released: {}", taskGroupQueue.getId());
}
@Override
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSubmittedStateAction.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSubmittedStateAction.java
index 22c71cc1d7..05b98f93c7 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSubmittedStateAction.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSubmittedStateAction.java
@@ -143,6 +143,12 @@ public class TaskSubmittedStateAction extends AbstractTaskStateAction {
taskExecution.getWorkflowEventBus().publish(TaskPausedLifecycleEvent.of(taskExecution));
return;
}
+ if (taskExecution.getTaskInstance().getTaskGroupId() > 0
+ && taskGroupCoordinator.releaseWaitingTaskGroupSlot(taskExecution.getTaskInstance())) {
+ log.info("Success pause task: {} while waiting for TaskGroup slot", taskExecution.getName());
+ taskExecution.getWorkflowEventBus().publish(TaskPausedLifecycleEvent.of(taskExecution));
+ return;
+ }
log.info("The task[id={}] is submitted and already dispatched, cannot pause, will try to pause it after 5s",
taskExecution.getId());
taskExecution.getWorkflowEventBus()
@@ -167,6 +173,12 @@ public class TaskSubmittedStateAction extends AbstractTaskStateAction {
taskExecution.getWorkflowEventBus().publish(TaskKilledLifecycleEvent.of(taskExecution));
return;
}
+ if (taskExecution.getTaskInstance().getTaskGroupId() > 0
+ && taskGroupCoordinator.releaseWaitingTaskGroupSlot(taskExecution.getTaskInstance())) {
+ log.info("Success kill task: {} while waiting for TaskGroup slot", taskExecution.getName());
+ taskExecution.getWorkflowEventBus().publish(TaskKilledLifecycleEvent.of(taskExecution));
+ return;
+ }
log.info("The task[id={}] is submitted and already dispatched, cannot kill, will kill it after 5s",
taskExecution.getId());
taskExecution.getWorkflowEventBus()
diff --git a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartTaskGroupTestCase.java b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartTaskGroupTestCase.java
index 4fb7bc5937..18388aa457 100644
--- a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartTaskGroupTestCase.java
+++ b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartTaskGroupTestCase.java
@@ -19,8 +19,16 @@ package org.apache.dolphinscheduler.server.master.integration.cases;
import static org.awaitility.Awaitility.await;
+import org.apache.dolphinscheduler.common.enums.Flag;
+import org.apache.dolphinscheduler.common.enums.TaskGroupQueueStatus;
+import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
+import org.apache.dolphinscheduler.dao.entity.TaskGroup;
+import org.apache.dolphinscheduler.dao.entity.TaskGroupQueue;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
+import org.apache.dolphinscheduler.dao.repository.TaskGroupDao;
+import org.apache.dolphinscheduler.dao.repository.TaskGroupQueueDao;
+import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.extract.master.command.RunWorkflowCommandParam;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import org.apache.dolphinscheduler.server.master.AbstractMasterIntegrationTestCase;
@@ -28,17 +36,28 @@ import org.apache.dolphinscheduler.server.master.integration.WorkflowOperator;
import org.apache.dolphinscheduler.server.master.integration.WorkflowTestCaseContext;
import java.time.Duration;
+import java.util.Date;
import java.util.List;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
/**
* Integration tests for workflow start task group scenarios.
*/
public class WorkflowStartTaskGroupTestCase extends AbstractMasterIntegrationTestCase {
+ @Autowired
+ private TaskGroupDao taskGroupDao;
+
+ @Autowired
+ private TaskGroupQueueDao taskGroupQueueDao;
+
+ @Autowired
+ private TaskInstanceDao taskInstanceDao;
+
@Test
@DisplayName("Test start a workflow with two fake task(A) using task group")
public void testStartWorkflow_with_successTaskUsingTaskGroup() {
@@ -79,4 +98,114 @@ public class WorkflowStartTaskGroupTestCase extends AbstractMasterIntegrationTes
masterContainer.assertAllResourceReleased();
}
+ @Test
+ @DisplayName("Test pause a workflow while tasks are waiting for a task group slot")
+ public void testPauseWorkflow_withTasksWaitingForTaskGroupSlot() {
+ final Integer workflowInstanceId = triggerWorkflowWithTasksWaitingForTaskGroupSlot();
+
+ Assertions.assertThat(workflowOperator.pauseWorkflowInstance(workflowInstanceId).isSuccess()).isTrue();
+
+ assertWorkflowAndTasksFinished(workflowInstanceId, WorkflowExecutionStatus.PAUSE, TaskExecutionStatus.PAUSE);
+ masterContainer.assertAllResourceReleased();
+ }
+
+ @Test
+ @DisplayName("Test stop a workflow while tasks are waiting for a task group slot")
+ public void testStopWorkflow_withTasksWaitingForTaskGroupSlot() {
+ final Integer workflowInstanceId = triggerWorkflowWithTasksWaitingForTaskGroupSlot();
+
+ Assertions.assertThat(workflowOperator.stopWorkflowInstance(workflowInstanceId).isSuccess()).isTrue();
+
+ assertWorkflowAndTasksFinished(workflowInstanceId, WorkflowExecutionStatus.STOP, TaskExecutionStatus.KILL);
+ masterContainer.assertAllResourceReleased();
+ }
+
+ private Integer triggerWorkflowWithTasksWaitingForTaskGroupSlot() {
+ final String yaml = "/it/start/workflow_with_fake_tasks_using_task_group.yaml";
+ final WorkflowTestCaseContext context = workflowTestCaseContextFactory.initializeContextFromYaml(yaml);
+ final WorkflowDefinition workflow = context.getOneWorkflow();
+ final TaskGroup taskGroup = context.getTaskGroups().get(0);
+ occupyTaskGroupSlot(taskGroup);
+
+ final WorkflowOperator.WorkflowTriggerDTO workflowTriggerDTO = WorkflowOperator.WorkflowTriggerDTO.builder()
+ .workflowDefinition(workflow)
+ .runWorkflowCommandParam(new RunWorkflowCommandParam())
+ .build();
+ final Integer workflowInstanceId = workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
+
+ await()
+ .pollInterval(Duration.ofMillis(100))
+ .atMost(Duration.ofMinutes(1))
+ .untilAsserted(() -> {
+ Assertions.assertThat(repository.queryWorkflowInstance(workflowInstanceId).getState())
+ .isEqualTo(WorkflowExecutionStatus.RUNNING_EXECUTION);
+ Assertions.assertThat(repository.queryTaskInstance(workflowInstanceId))
+ .hasSize(2)
+ .allSatisfy(taskInstance -> {
+ Assertions.assertThat(taskInstance.getState())
+ .isEqualTo(TaskExecutionStatus.SUBMITTED_SUCCESS);
+ Assertions.assertThat(taskInstance.getTaskGroupId()).isEqualTo(taskGroup.getId());
+ Assertions.assertThat(taskGroupQueueDao.queryByTaskInstanceId(taskInstance.getId()))
+ .singleElement()
+ .extracting(TaskGroupQueue::getStatus)
+ .isEqualTo(TaskGroupQueueStatus.WAIT_QUEUE);
+ });
+ });
+ return workflowInstanceId;
+ }
+
+ private void assertWorkflowAndTasksFinished(final Integer workflowInstanceId,
+ final WorkflowExecutionStatus workflowExecutionStatus,
+ final TaskExecutionStatus taskExecutionStatus) {
+ await()
+ .pollInterval(Duration.ofMillis(100))
+ .atMost(Duration.ofMinutes(1))
+ .untilAsserted(() -> {
+ Assertions.assertThat(repository.queryWorkflowInstance(workflowInstanceId).getState())
+ .isEqualTo(workflowExecutionStatus);
+ Assertions.assertThat(repository.queryTaskInstance(workflowInstanceId))
+ .hasSize(2)
+ .allSatisfy(taskInstance -> {
+ Assertions.assertThat(taskInstance.getState()).isEqualTo(taskExecutionStatus);
+ Assertions.assertThat(taskGroupQueueDao.queryByTaskInstanceId(taskInstance.getId()))
+ .isEmpty();
+ });
+ });
+ }
+
+ private void occupyTaskGroupSlot(final TaskGroup taskGroup) {
+ final Date now = new Date();
+ final TaskInstance slotHolder = TaskInstance.builder()
+ .name("task-group-slot-holder")
+ .taskType("LogicFakeTask")
+ .workflowInstanceId(0)
+ .workflowInstanceName("task-group-slot-holder")
+ .taskCode(Long.MAX_VALUE)
+ .taskDefinitionVersion(1)
+ .state(TaskExecutionStatus.RUNNING_EXECUTION)
+ .flag(Flag.YES)
+ .submitTime(now)
+ .firstSubmitTime(now)
+ .startTime(now)
+ .taskGroupId(taskGroup.getId())
+ .build();
+ Assertions.assertThat(taskInstanceDao.insert(slotHolder)).isEqualTo(1);
+
+ final TaskGroupQueue slotHolderQueue = TaskGroupQueue.builder()
+ .taskId(slotHolder.getId())
+ .taskName(slotHolder.getName())
+ .groupId(taskGroup.getId())
+ .priority(0)
+ .forceStart(Flag.NO.getCode())
+ .inQueue(Flag.YES.getCode())
+ .status(TaskGroupQueueStatus.ACQUIRE_SUCCESS)
+ .createTime(now)
+ .updateTime(now)
+ .build();
+ Assertions.assertThat(taskGroupQueueDao.insert(slotHolderQueue)).isEqualTo(1);
+
+ taskGroup.setUseSize(taskGroup.getGroupSize());
+ Assertions.assertThat(taskGroupDao.updateById(taskGroup)).isTrue();
+ }
+
}