Unable to retrieve from queue after completing workflow

**Describe the bug**
Unable to retrieve from queue after completing workflow

**Details**
Conductor version: laster

`
private void endExecution(WorkflowModel workflow, TaskModel terminateTask) {
if (terminateTask != null) {
String terminationStatus =
(String)
terminateTask
.getInputData()
.get(Terminate.getTerminationStatusParameter());
String reason =
(String)
terminateTask
.getInputData()
.get(Terminate.getTerminationReasonParameter());
if (StringUtils.isBlank(reason)) {
reason =
String.format(
"Workflow is %s by TERMINATE task: %s",
terminationStatus, terminateTask.getTaskId());
}
if (WorkflowModel.Status.FAILED.name().equals(terminationStatus)) {
workflow.setStatus(WorkflowModel.Status.FAILED);
workflow =
terminate(
workflow,
new TerminateWorkflowException(
reason, workflow.getStatus(), terminateTask));
} else {
workflow.setReasonForIncompletion(reason);
workflow = completeWorkflow(workflow);
}
} else {
workflow = completeWorkflow(workflow);
}

cancelNonTerminalTasks(workflow);
}
WorkflowModel completeWorkflow(WorkflowModel workflow) {
LOGGER.debug("Completing workflow execution for {}", workflow.getWorkflowId());

if (workflow.getStatus().equals(WorkflowModel.Status.COMPLETED)) {
queueDAO.remove(DECIDER_QUEUE, workflow.getWorkflowId()); // remove from the sweep queue
executionDAOFacade.removeFromPendingWorkflow(
workflow.getWorkflowName(), workflow.getWorkflowId());
LOGGER.debug("Workflow: {} has already been completed.", workflow.getWorkflowId());
return workflow;
}

if (workflow.getStatus().isTerminal()) {
String msg =
"Workflow is already in terminal state. Current status: "
+ workflow.getStatus();
throw new ConflictException(msg);
}

deciderService.updateWorkflowOutput(workflow, null);

workflow.setStatus(WorkflowModel.Status.COMPLETED);

// update the failed reference task names
List failedTasks =
workflow.getTasks().stream()
.filter(
t ->
FAILED.equals(t.getStatus())
|| FAILED_WITH_TERMINAL_ERROR.equals(t.getStatus()))
.collect(Collectors.toList());
workflow.getFailedReferenceTaskNames()
.addAll(
failedTasks.stream()
.map(TaskModel::getReferenceTaskName)
.collect(Collectors.toSet()));

workflow.getFailedTaskNames()
.addAll(
failedTasks.stream()
.map(TaskModel::getTaskDefName)
.collect(Collectors.toSet()));
executionDAOFacade.updateWorkflow(workflow);
LOGGER.debug("Completed workflow execution for {}", workflow.getWorkflowId());
//通知任务状态完成
workflowStatusListener.onWorkflowCompletedIfEnabled(workflow);
Monitors.recordWorkflowCompletion(
workflow.getWorkflowName(),
workflow.getEndTime() - workflow.getCreateTime(),
workflow.getOwnerApp());

if (workflow.hasParent()) {
updateParentWorkflowTask(workflow);
LOGGER.info(
"{} updated parent {} task {}",
workflow.toShortString(),
workflow.getParentWorkflowId(),
workflow.getParentWorkflowTaskId());
expediteLazyWorkflowEvaluation(workflow.getParentWorkflowId());
}

executionLockService.releaseLock(workflow.getWorkflowId());
executionLockService.deleteLock(workflow.getWorkflowId());
return workflow;
}
`

After the `completeWorkflow` method is executed, the following logic is not executed. I understand that there is a missing callback, which may be a problem here!
`
if (workflow.getStatus().equals(WorkflowModel.Status.COMPLETED)) {
queueDAO.remove(DECIDER_QUEUE, workflow.getWorkflowId()); // remove from the sweep queue
executionDAOFacade.removeFromPendingWorkflow(
workflow.getWorkflowName(), workflow.getWorkflowId());
LOGGER.debug("Workflow: {} has already been completed.", workflow.getWorkflowId());
return workflow;
}
`

I think you may need to re-execute completeWorkflow before `com.netflix.conductor.core.execution.WorkflowExecutorOps#endExecution#cancelNonTerminalTasks` to properly clean up the sweep queue and workflow