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

Repair inconsistent subworkflow task #3183

Merged
merged 1 commit into from
Aug 16, 2022
Merged

Repair inconsistent subworkflow task #3183

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
26 changes: 26 additions & 0 deletions core/src/main/java/com/netflix/conductor/core/reconciliation/WorkflowRepairService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import org.springframework.stereotype.Service;

import com.netflix.conductor.annotations.VisibleForTesting;
import com.netflix.conductor.common.metadata.tasks.TaskType;
import com.netflix.conductor.core.config.ConductorProperties;
import com.netflix.conductor.core.execution.tasks.SystemTaskRegistry;
import com.netflix.conductor.core.execution.tasks.WorkflowSystemTask;
Expand Down Expand Up @@ -138,6 +139,13 @@ boolean verifyAndRepairTask(TaskModel task) {
Monitors.recordQueueMessageRepushFromRepairService(task.getTaskDefName());
return true;
}
} else if (task.getTaskType().equals(TaskType.TASK_TYPE_SUB_WORKFLOW)
&& task.getStatus() == TaskModel.Status.IN_PROGRESS) {
WorkflowModel subWorkflow = executionDAO.getWorkflow(task.getSubWorkflowId(), false);
if (subWorkflow.getStatus().isTerminal()) {
repairSubWorkflowTask(task, subWorkflow.getStatus());
return true;
}
}
return false;
}
Expand All @@ -156,4 +164,22 @@ private boolean verifyAndRepairWorkflow(String workflowId) {
}
return false;
}

private void repairSubWorkflowTask(TaskModel task, WorkflowModel.Status subWorkflowStatus) {
switch (subWorkflowStatus) {
case COMPLETED:
task.setStatus(TaskModel.Status.COMPLETED);
break;
case FAILED:
task.setStatus(TaskModel.Status.FAILED);
break;
case TERMINATED:
task.setStatus(TaskModel.Status.CANCELED);
break;
case TIMED_OUT:
task.setStatus(TaskModel.Status.TIMED_OUT);
break;
}
executionDAO.updateTask(task);
}
}
48 changes: 43 additions & 5 deletions core/src/test/java/com/netflix/conductor/core/reconciliation/TestWorkflowRepairService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,13 +13,17 @@
package com.netflix.conductor.core.reconciliation;

import java.time.Duration;
import java.util.Map;

import org.junit.Before;
import org.junit.Test;
import org.mockito.ArgumentCaptor;

import com.netflix.conductor.core.config.ConductorProperties;
import com.netflix.conductor.core.events.EventQueues;
import com.netflix.conductor.core.execution.WorkflowExecutor;
import com.netflix.conductor.core.execution.tasks.*;
import com.netflix.conductor.core.utils.ParametersUtils;
import com.netflix.conductor.dao.ExecutionDAO;
import com.netflix.conductor.dao.QueueDAO;
import com.netflix.conductor.model.TaskModel;
Expand All @@ -29,8 +33,7 @@

import static com.netflix.conductor.common.metadata.tasks.TaskType.*;

import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.*;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
Expand Down Expand Up @@ -169,13 +172,18 @@ public void assertSyncSystemTasksAreNotCheckedAgainstQueue() {
@Test
public void assertAsyncCompleteInProgressSystemTasksAreNotCheckedAgainstQueue() {
TaskModel task = new TaskModel();
task.setTaskType(TASK_TYPE_SUB_WORKFLOW);
task.setTaskType(TASK_TYPE_EVENT);
task.setStatus(TaskModel.Status.IN_PROGRESS);
task.setTaskId("abcd");
task.setCallbackAfterSeconds(60);
task.setInputData(Map.of("asyncComplete", true));

WorkflowSystemTask workflowSystemTask = new SubWorkflow(new ObjectMapper());
when(systemTaskRegistry.get(TASK_TYPE_SUB_WORKFLOW)).thenReturn(workflowSystemTask);
WorkflowSystemTask workflowSystemTask =
new Event(
mock(EventQueues.class),
mock(ParametersUtils.class),
mock(ObjectMapper.class));
when(systemTaskRegistry.get(TASK_TYPE_EVENT)).thenReturn(workflowSystemTask);

assertTrue(workflowSystemTask.isAsyncComplete(task));

Expand Down Expand Up @@ -219,4 +227,34 @@ public void verifyAndRepairParentWorkflow() {
verify(queueDAO, times(1)).containsMessage(anyString(), anyString());
verify(queueDAO, times(1)).push(anyString(), anyString(), anyLong());
}

@Test
public void assertInProgressSubWorkflowSystemTasksAreCheckedAndRepaired() {
String subWorkflowId = "subWorkflowId";
String taskId = "taskId";

TaskModel task = new TaskModel();
task.setTaskType(TASK_TYPE_SUB_WORKFLOW);
task.setStatus(TaskModel.Status.IN_PROGRESS);
task.setTaskId(taskId);
task.setCallbackAfterSeconds(60);
task.setSubWorkflowId(subWorkflowId);

WorkflowModel subWorkflow = new WorkflowModel();
subWorkflow.setWorkflowId(subWorkflowId);
subWorkflow.setStatus(WorkflowModel.Status.TERMINATED);

when(executionDAO.getWorkflow(subWorkflowId, false)).thenReturn(subWorkflow);

assertTrue(workflowRepairService.verifyAndRepairTask(task));
// Verify that queue message is never pushed for async complete system tasks
verify(queueDAO, never()).containsMessage(anyString(), anyString());
verify(queueDAO, never()).push(anyString(), anyString(), anyLong());
// Verify
ArgumentCaptor<TaskModel> argumentCaptor = ArgumentCaptor.forClass(TaskModel.class);
verify(executionDAO, times(1)).updateTask(argumentCaptor.capture());
assertEquals(taskId, argumentCaptor.getValue().getTaskId());
assertEquals(subWorkflowId, argumentCaptor.getValue().getSubWorkflowId());
assertEquals(TaskModel.Status.CANCELED, argumentCaptor.getValue().getStatus());
}
}