Skip to content

Ensure index operations succeed before removing workflows from ExecutionDAO #709

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rajeshwar-nu wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Ensure index operations succeed before removing workflows from ExecutionDAO #709

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
14d69b3
Fixes workflow removal sequencing
rajeshwar-nu Jan 8, 2026
254c7ff
Do not fail when workflow/task not found on non-archival cases
rajeshwar-nu Jan 8, 2026
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
41 changes: 38 additions & 3 deletions core/src/main/java/com/netflix/conductor/core/dal/ExecutionDAOFacade.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -333,31 +333,57 @@ public void removeFromPendingWorkflow(String workflowType, String workflowId) {
*
* @param workflowId the id of the workflow to be removed
* @param archiveWorkflow if true, the workflow and associated tasks will be archived in the
* {@link IndexDAO} after removal from {@link ExecutionDAO}.
* {@link IndexDAO} before removal from {@link ExecutionDAO}.
*/
public void removeWorkflow(String workflowId, boolean archiveWorkflow) {
WorkflowModel workflow = getWorkflowModelFromDataStore(workflowId, true);

executionDAO.removeWorkflow(workflowId);
// Index operations happen before DAO removal to prevent data loss on index failures.
try {
removeWorkflowIndex(workflow, archiveWorkflow);
} catch (NotFoundException e) {
if (archiveWorkflow) {
throw e;
}
// Idempotent deletion: missing index records should not block DAO removal.
LOGGER.info("Workflow {} not found in index during removal, continuing", workflowId, e);
} catch (JsonProcessingException e) {
throw new TransientException("Workflow can not be serialized to json", e);
}

// Task index removals run before DAO deletion for the same consistency guarantees.
workflow.getTasks()
.forEach(
task -> {
try {
removeTaskIndex(workflow, task, archiveWorkflow);
} catch (NotFoundException e) {
if (archiveWorkflow) {
throw e;
}
// Idempotent deletion: missing index records should not block DAO
// removal.
LOGGER.info(
"Task {} of workflow {} not found in index during removal, continuing",
task.getTaskId(),
workflowId,
e);
} catch (JsonProcessingException e) {
throw new TransientException(
String.format(
"Task %s of workflow %s can not be serialized to json",
task.getTaskId(), workflow.getWorkflowId()),
e);
}
});

// Only remove from the source of truth after index operations succeed.
executionDAO.removeWorkflow(workflowId);

// finally remove from queues
workflow.getTasks()
.forEach(
task -> {
try {
queueDAO.remove(QueueUtils.getQueueName(task), task.getTaskId());
} catch (Exception e) {
Expand Down Expand Up @@ -404,7 +430,16 @@ public void removeWorkflowWithExpiry(
try {
WorkflowModel workflow = getWorkflowModelFromDataStore(workflowId, true);

removeWorkflowIndex(workflow, archiveWorkflow);
try {
removeWorkflowIndex(workflow, archiveWorkflow);
} catch (NotFoundException e) {
if (archiveWorkflow) {
throw e;
}
// Idempotent deletion: missing index records should not block DAO removal.
LOGGER.info(
"Workflow {} not found in index during removal, continuing", workflowId, e);
}
// remove workflow from DAO with TTL
executionDAO.removeWorkflowWithExpiry(workflowId, ttlSeconds);
} catch (Exception e) {
Expand Down
106 changes: 106 additions & 0 deletions core/src/test/java/com/netflix/conductor/core/dal/ExecutionDAOFacadeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,10 +28,12 @@

import com.netflix.conductor.common.config.TestObjectMapperConfiguration;
import com.netflix.conductor.common.metadata.events.EventExecution;
import com.netflix.conductor.common.metadata.workflow.WorkflowDef;
import com.netflix.conductor.common.run.SearchResult;
import com.netflix.conductor.common.run.Workflow;
import com.netflix.conductor.common.utils.ExternalPayloadStorage;
import com.netflix.conductor.core.config.ConductorProperties;
import com.netflix.conductor.core.exception.NotFoundException;
import com.netflix.conductor.core.exception.TerminateWorkflowException;
import com.netflix.conductor.core.execution.TestDeciderService;
import com.netflix.conductor.core.utils.ExternalPayloadStorageUtils;
Expand Down Expand Up @@ -156,6 +158,46 @@ public void testRemoveWorkflow() {
verify(indexDAO, times(1)).asyncRemoveTask(anyString(), anyString());
}

@Test
public void testRemoveWorkflowContinuesOnIndexNotFound() {
WorkflowModel workflow = new WorkflowModel();
workflow.setWorkflowId("workflowId");
workflow.setStatus(WorkflowModel.Status.COMPLETED);

TaskModel task = new TaskModel();
task.setTaskId("taskId");
workflow.setTasks(Collections.singletonList(task));

when(executionDAO.getWorkflow(anyString(), anyBoolean())).thenReturn(workflow);
doThrow(new NotFoundException("missing workflow"))
.when(indexDAO)
.asyncRemoveWorkflow(anyString());

executionDAOFacade.removeWorkflow("workflowId", false);

verify(executionDAO, times(1)).removeWorkflow(anyString());
}

@Test
public void testRemoveWorkflowContinuesOnTaskIndexNotFound() {
WorkflowModel workflow = new WorkflowModel();
workflow.setWorkflowId("workflowId");
workflow.setStatus(WorkflowModel.Status.COMPLETED);

TaskModel task = new TaskModel();
task.setTaskId("taskId");
workflow.setTasks(Collections.singletonList(task));

when(executionDAO.getWorkflow(anyString(), anyBoolean())).thenReturn(workflow);
doThrow(new NotFoundException("missing task"))
.when(indexDAO)
.asyncRemoveTask(anyString(), anyString());

executionDAOFacade.removeWorkflow("workflowId", false);

verify(executionDAO, times(1)).removeWorkflow(anyString());
}

@Test
public void testArchiveWorkflow() throws Exception {
InputStream stream = TestDeciderService.class.getResourceAsStream("/completed.json");
Expand All @@ -171,6 +213,70 @@ public void testArchiveWorkflow() throws Exception {
verify(indexDAO, never()).removeTask(anyString(), anyString());
}

@Test
public void testArchiveWorkflowSkipsRemovalOnIndexFailure() throws Exception {
InputStream stream = TestDeciderService.class.getResourceAsStream("/completed.json");
WorkflowModel workflow = objectMapper.readValue(stream, WorkflowModel.class);

when(executionDAO.getWorkflow(anyString(), anyBoolean())).thenReturn(workflow);
doThrow(new RuntimeException("index failure"))
.when(indexDAO)
.updateWorkflow(anyString(), any(), any());

assertThrows(
RuntimeException.class,
() -> executionDAOFacade.removeWorkflow("workflowId", true));

verify(executionDAO, never()).removeWorkflow(anyString());
}

@Test
public void testArchiveWorkflowSkipsRemovalOnTaskArchiveFailure() {
WorkflowModel workflow = new WorkflowModel();
WorkflowDef workflowDef = new WorkflowDef();
workflowDef.setName("workflowName");
workflowDef.setVersion(1);
workflow.setWorkflowDefinition(workflowDef);
workflow.setWorkflowId("workflowId");
workflow.setStatus(WorkflowModel.Status.COMPLETED);

TaskModel task = new TaskModel();
task.setTaskId("taskId");
task.setStatus(TaskModel.Status.IN_PROGRESS);
workflow.setTasks(Collections.singletonList(task));

when(executionDAO.getWorkflow(anyString(), anyBoolean())).thenReturn(workflow);

assertThrows(
IllegalArgumentException.class,
() -> executionDAOFacade.removeWorkflow("workflowId", true));

verify(indexDAO, times(1)).updateWorkflow(anyString(), any(), any());
verify(executionDAO, never()).removeWorkflow(anyString());
}

@Test
public void testRemoveWorkflowSkipsRemovalOnIndexFailure() {
WorkflowModel workflow = new WorkflowModel();
workflow.setWorkflowId("workflowId");
workflow.setStatus(WorkflowModel.Status.COMPLETED);

TaskModel task = new TaskModel();
task.setTaskId("taskId");
workflow.setTasks(Collections.singletonList(task));

when(executionDAO.getWorkflow(anyString(), anyBoolean())).thenReturn(workflow);
doThrow(new RuntimeException("index failure"))
.when(indexDAO)
.asyncRemoveWorkflow(anyString());

assertThrows(
RuntimeException.class,
() -> executionDAOFacade.removeWorkflow("workflowId", false));

verify(executionDAO, never()).removeWorkflow(anyString());
}

@Test
public void testAddEventExecution() {
when(executionDAO.addEventExecution(any())).thenReturn(false);
Expand Down