Skip to content
This repository has been archived by the owner on Dec 13, 2023. It is now read-only.

Commit

Permalink
Merge pull request #3639 from yohanyflores/fix-task-not-working-properly
Browse files Browse the repository at this point in the history 
Fix WAIT task is not working properly (after v3.13.0)
  • Loading branch information
@v1r3n
v1r3n authored Jun 19, 2023
2 parents e68dee6 + bcf9641 commit 3920aee
Show file tree
Hide file tree
Showing 2 changed files with 70 additions and 9 deletions.
16 changes: 10 additions & 6 deletions core/src/main/java/com/netflix/conductor/core/reconciliation/WorkflowSweeper.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -117,12 +117,16 @@ void unack(WorkflowModel workflowModel, long workflowOffsetTimeout) {
long postponeDurationSeconds = 0;
for (TaskModel taskModel : workflowModel.getTasks()) {
if (taskModel.getStatus() == Status.IN_PROGRESS) {
if (taskModel.getTaskType().equals(TaskType.TASK_TYPE_WAIT)
|| taskModel.getTaskType().equals(TaskType.TASK_TYPE_HUMAN)) {
postponeDurationSeconds =
(taskModel.getWaitTimeout() != 0)
? taskModel.getWaitTimeout() + 1
: workflowOffsetTimeout;
if (taskModel.getTaskType().equals(TaskType.TASK_TYPE_WAIT)) {
if (taskModel.getWaitTimeout() == 0) {
postponeDurationSeconds = workflowOffsetTimeout;
} else {
long deltaInSeconds =
(taskModel.getWaitTimeout() - System.currentTimeMillis()) / 1000;
postponeDurationSeconds = (deltaInSeconds > 0) ? deltaInSeconds : 0;
}
} else if (taskModel.getTaskType().equals(TaskType.TASK_TYPE_HUMAN)) {
postponeDurationSeconds = workflowOffsetTimeout;
} else {
postponeDurationSeconds =
(taskModel.getResponseTimeoutSeconds() != 0)
Expand Down
63 changes: 60 additions & 3 deletions core/src/test/java/com/netflix/conductor/core/reconciliation/TestWorkflowSweeper.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,25 @@ public void setUp() {
workflowExecutor, Optional.of(workflowRepairService), properties, queueDAO);
}

@Test
public void testPostponeDurationForHumanTaskType() {
WorkflowModel workflowModel = new WorkflowModel();
workflowModel.setWorkflowId("1");
TaskModel taskModel = new TaskModel();
taskModel.setTaskId("task1");
taskModel.setTaskType(TaskType.TASK_TYPE_HUMAN);
taskModel.setStatus(Status.IN_PROGRESS);
workflowModel.setTasks(List.of(taskModel));
when(properties.getWorkflowOffsetTimeout())
.thenReturn(Duration.ofSeconds(defaultPostPoneOffSetSeconds));
workflowSweeper.unack(workflowModel, defaultPostPoneOffSetSeconds);
verify(queueDAO)
.setUnackTimeout(
DECIDER_QUEUE,
workflowModel.getWorkflowId(),
defaultPostPoneOffSetSeconds * 1000);
}

@Test
public void testPostponeDurationForWaitTaskType() {
WorkflowModel workflowModel = new WorkflowModel();
Expand All @@ -77,22 +96,60 @@ public void testPostponeDurationForWaitTaskType() {
}

@Test
public void testPostponeDurationForWaitTaskTypeWithWaitTime() {
public void testPostponeDurationForWaitTaskTypeWithLongWaitTime() {
long waitTimeout = 65845;
WorkflowModel workflowModel = new WorkflowModel();
workflowModel.setWorkflowId("1");
TaskModel taskModel = new TaskModel();
taskModel.setTaskId("task1");
taskModel.setTaskType(TaskType.TASK_TYPE_WAIT);
taskModel.setStatus(Status.IN_PROGRESS);
taskModel.setWaitTimeout(System.currentTimeMillis() + waitTimeout);
workflowModel.setTasks(List.of(taskModel));
when(properties.getWorkflowOffsetTimeout())
.thenReturn(Duration.ofSeconds(defaultPostPoneOffSetSeconds));
workflowSweeper.unack(workflowModel, defaultPostPoneOffSetSeconds);
verify(queueDAO)
.setUnackTimeout(
DECIDER_QUEUE, workflowModel.getWorkflowId(), (waitTimeout / 1000) * 1000);
}

@Test
public void testPostponeDurationForWaitTaskTypeWithLessOneSecondWaitTime() {
long waitTimeout = 180;
WorkflowModel workflowModel = new WorkflowModel();
workflowModel.setWorkflowId("1");
TaskModel taskModel = new TaskModel();
taskModel.setTaskId("task1");
taskModel.setTaskType(TaskType.TASK_TYPE_WAIT);
taskModel.setStatus(Status.IN_PROGRESS);
taskModel.setWaitTimeout(waitTimeout);
taskModel.setWaitTimeout(System.currentTimeMillis() + waitTimeout);
workflowModel.setTasks(List.of(taskModel));
when(properties.getWorkflowOffsetTimeout())
.thenReturn(Duration.ofSeconds(defaultPostPoneOffSetSeconds));
workflowSweeper.unack(workflowModel, defaultPostPoneOffSetSeconds);
verify(queueDAO)
.setUnackTimeout(
DECIDER_QUEUE, workflowModel.getWorkflowId(), (waitTimeout / 1000) * 1000);
}

@Test
public void testPostponeDurationForWaitTaskTypeWithZeroWaitTime() {
long waitTimeout = 0;
WorkflowModel workflowModel = new WorkflowModel();
workflowModel.setWorkflowId("1");
TaskModel taskModel = new TaskModel();
taskModel.setTaskId("task1");
taskModel.setTaskType(TaskType.TASK_TYPE_WAIT);
taskModel.setStatus(Status.IN_PROGRESS);
taskModel.setWaitTimeout(System.currentTimeMillis() + waitTimeout);
workflowModel.setTasks(List.of(taskModel));
when(properties.getWorkflowOffsetTimeout())
.thenReturn(Duration.ofSeconds(defaultPostPoneOffSetSeconds));
workflowSweeper.unack(workflowModel, defaultPostPoneOffSetSeconds);
verify(queueDAO)
.setUnackTimeout(
DECIDER_QUEUE, workflowModel.getWorkflowId(), (waitTimeout + 1) * 1000);
DECIDER_QUEUE, workflowModel.getWorkflowId(), (waitTimeout / 1000) * 1000);
}

@Test
Expand Down

0 comments on commit 3920aee

Please sign in to comment.