WorkflowRepairService { @VisibleForTesting protected boolean verifyAndRepairTask(Task task) { WorkflowSystemTask workflowSystemTask = WorkflowSystemTask.get(task.getTaskType()); if (task.getStatus() == Task.Status.SCHEDULED) { if (isSystemTask.test(task) && !workflowSystemTask.isAsync()) { return false; } String taskQueueName = QueueUtils.getQueueName(task); if (!queueDAO.containsMessage(taskQueueName, task.getTaskId())) { queueDAO.push(taskQueueName, task.getTaskId(), task.getCallbackAfterSeconds()); Monitors.recordQueueMessageRepushFromRepairService(task.getTaskDefName()); return true; } } return false; } @Inject  WorkflowRepairService(
            ExecutionDAO executionDAO,
            QueueDAO queueDAO,
            Configuration configuration
    ); boolean verifyAndRepairWorkflow(String workflowId, boolean includeTasks); void verifyAndRepairWorkflowTasks(String workflowId);  }
@Test public void verifyAndRepairTask() { Task task = new Task(); task.setTaskType("SIMPLE"); task.setStatus(Task.Status.SCHEDULED); task.setTaskId("abcd"); task.setCallbackAfterSeconds(60); when(queueDAO.containsMessage(anyString(), anyString())).thenReturn(false); assertTrue(workflowRepairService.verifyAndRepairTask(task)); verify(queueDAO, times(1)).push(anyString(), anyString(), anyLong()); }
@Test public void assertSyncSystemTasksAreNotCheckedAgainstQueue() { Decision decision = new Decision(); Task task = new Task(); task.setTaskType("DECISION"); task.setStatus(Task.Status.SCHEDULED); assertFalse(workflowRepairService.verifyAndRepairTask(task)); verify(queueDAO, never()).containsMessage(anyString(), anyString()); verify(queueDAO, never()).push(anyString(), anyString(), anyLong()); }