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

Fix WAIT task is not working properly (after v3.13.0) #3639

Merged
merged 6 commits into from
Jun 19, 2023
+70 −9
Merged

Fix WAIT task is not working properly (after v3.13.0) #3639

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
3d02f22
Assign the value in seconds to postponeDurationSeconds in WAIT
yohanyflores Jun 4, 2023
dc798dd
Fix Fix WAIT task is not working properly (after v3.13.0) #3402
yohanyflores Jun 5, 2023
ca6717e
Support Wait with waittimeout = 0
yohanyflores Jun 9, 2023
ca1830c
Merge branch 'main' into fix-task-not-working-properly
yohanyflores Jun 10, 2023
5d32287
Fix Wait one second more
yohanyflores Jun 15, 2023
bcf9641
Refactoring Long Literal
yohanyflores Jun 15, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 10 additions & 6 deletions core/src/main/java/com/netflix/conductor/core/reconciliation/WorkflowSweeper.java
View file
Open in desktop
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
Open in desktop
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