asyncComplete is specified to true，however simple task is executed multiple times。

[zhouwanbin]

Example workflow:
{
"name": "json-worker-flow",
"description": "json-worker-flow",
"version": 0,
"tasks": [
{
"name": "taskC",
"taskReferenceName": "dt_taskD_0_ref",
"description": "dt_taskD_0",
"inputParameters": {
"city": "${workflow.input.city}"
},
"startDelay": 0,
"asyncComplete": true,
"retryCount": 0,
"workflowTaskType": "SIMPLE"
}
],
"timeoutPolicy": "TIME_OUT_WF",
"timeoutSeconds": 3600,
"ownerEmail": "[email protected]"
}

Example task：
public class AsyncWorkerC implements Worker {

private final String taskDefName = "taskC";

@Override
public String getTaskDefName() {
    return taskDefName;
}

private Set<String> taskIdSet = new HashSet<>();


@Override
public TaskResult execute(Task task) {
    LogUtils.printTask(this.getClass(), task);
    TaskResult result = new TaskResult(task);
    //result.setStatus(Status.IN_PROGRESS);
    if (!taskIdSet.contains(task.getTaskId())) {
        taskIdSet.add(task.getTaskId());
        LogUtils.info("AsyncWorkerC execute real task = " + task);
        asyncTask(task);
    } else {
        LogUtils.info("AsyncWorkerC execute repeat task = " + task);
    }
    return result;
}


private void asyncTask(Task task) {
    new Thread(() -> {
        LogUtils.info("AsyncWorkerC pre asyncTask task = " + task);
        try {
            Thread.sleep(300 * 1000);
        } catch (Exception e) {

        }
        LogUtils.info("AsyncWorkerC begin asyncTask task = " + task);
        TaskResult result = new TaskResult(task);
        result.setStatus(Status.COMPLETED);
        result.getOutputData().put("outputKey", "taskC-OutputKey2");
        result.getOutputData().put("oddEven", "taskC-Even2");
        result.getOutputData().put("mod", "taskC-Mod2");
        WorkerFlow.getInstance().getTaskClient().updateTask(result);
        LogUtils.info("AsyncWorkerC finish asyncTask task = " + task);
    }).start();
}

}

[BradenEads]

Hi @zhouwanbin,

In the provided code, you are using a Set taskIdSet to store the task IDs that have already been executed. You are checking whether the task ID is in the set before executing the task again, which helps to prevent the task from being executed multiple times.

However, this approach might not be sufficient to handle certain edge cases, especially in a distributed environment. To stop a simple task from being executed multiple times in a more reliable manner, consider the following suggestions:

1. Configure task retries in the workflow definition: If you are worried about the same task being retried multiple times due to failures, you can configure the retryCount and retryLogic in the workflow definition. For example, you can set retryCount to 1 to allow only one retry:

---

{
"name": "taskC",
...
"retryCount": 1,
...
}

2. Use a distributed locking mechanism: If your application is running in a distributed environment, you can use a distributed locking mechanism like Redis, ZooKeeper, or Consul to ensure that only one instance of the task is executed at a time. Acquire a lock using the task ID as the key before executing the task and release the lock once the task is completed or failed. This way, you can ensure that the task is not executed multiple times across different instances of your application.

3. Use Netflix Conductor's built-in deduplication feature: Starting from version 2.30.0, Netflix Conductor supports a built-in deduplication mechanism to prevent tasks from being executed multiple times. You can enable deduplication by setting the conductor.app.taskDeduplicationEnabled property to true in the Conductor configuration. Additionally, configure the deduplicationDelay property to define the delay in seconds before a duplicate task is allowed to be executed:

---

conductor:
app:
taskDeduplicationEnabled: true
deduplicationDelay: 60

By applying these suggestions, you can reduce the chances of a simple task being executed multiple times. Hope that's helpful!

[qingdaojunzuo]

Hi @BradenEads ,

We also encountered this issue these days. Any new way to cover this issue in latest version or asyncComplete supports SIMPLE task on framework level now?

We need,

• SIMPLE task only run once and never retry till the response received or timedout
• SIMPLE task waits in the queue in IN_PROGRESS status
• Mark SIMPLE task as SUCCESS/FAILED when external nofitication arrived
• The response received from external service can be added to the output of SIMPLE task
• The configuration can be set per task instance, setting in TASK CONFIGURATION rather than TASK DEF, such as, add when creating workflow definition meta in task list fields

---

• It seems that retryCount can be set on TASK CONFIGURATION level, but if retryCount set, the task status may be set to FAILED if count exceeded. We wants the status in IN_PROGRESS so that the left tasks are waiting for this status to be completed
• since run task once and never retry, deduplication function mentioned doesn't seem to meet this requrirement?

Currently, I tried 2 ways,

• Insert another WAIT task after current SIMPLE task, which means the SIMPLE task finished and the async notification complete/fail WAIT task and set the output of WAIT task
• Another way, it seems that callbackAfterSeconds in TaskResult can be set longer than task timeout, so that the task status will be SCHEDULED in queue after first run and then timedout to avoid next running. When nofitication arrived, we can complete/fail the SCHEDULED task directly

---

• For first way, I don't want to insert another system WAIT task based on previous task actions and complete/fail the result on WAIT task which is async result of previous SIMPLE task
• For second way, I found that, the timeout paramter is applied to TaskDef, not running WorkflowTask, if I have a general task, such as HTTP_WORKER, same function as system HTTP task, and when I need to do a kind of special call on fixed URL using HTTP_WORKER, I need to define a new WORKER(such as SPEC_HTTP_WORKER) to set the special timeout parameter on TaskDef or Can the task timeout be added when creating workflow definition in the task list fields?

Is there a formal way to do this now, the task requirement?

Thanks for your kindly help!