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 <[email protected]>
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();
+ }
+
}
