Do-While loop not working in v2.31.x

[ketantalreja]

Hi Team,

I am using Do-While loop task feature for looping over the tasks in my workflow and facing challenge while executing the workflow, do-while task remain in SCHEDULED state for some reason, not executing the loopOver tasks.

Here below is my workflow definition consisting of do-while task -

{
  "name": "DoWhile Test",
  "definition": {
    "updateTime": 1618590870149,
    "name": "DoWhile Test",
    "description": "do-while",
    "version": 6,
    "tasks": [
      {
        "name": "LoopTask",
        "taskReferenceName": "LoopTask",
        "inputParameters": {
          "value": 10
        },
        "type": "DOWHILE",
        "decisionCases": {},
        "defaultCase": [],
        "forkTasks": [],
        "startDelay": 0,
        "joinOn": [],
        "optional": false,
        "taskDefinition": {
          "name": "LoopTask",
          "retryCount": 3,
          "timeoutSeconds": 10000,
          "inputKeys": [],
          "outputKeys": [],
          "timeoutPolicy": "TIME_OUT_WF",
          "retryLogic": "FIXED",
          "retryDelaySeconds": 0,
          "responseTimeoutSeconds": 3600,
          "inputTemplate": {},
          "rateLimitPerFrequency": 0,
          "rateLimitFrequencyInSeconds": 1
        },
        "defaultExclusiveJoinTask": [],
        "asyncComplete": false,
        "loopCondition": "if ($.LoopTask['iteration'] < $.value) { true; } else { false; }",
        "loopOver": [
          {
            "name": "func_1",
            "taskReferenceName": "func_1",
            "inputParameters": {
              "scriptExpression": "$.LoopTask.output.iteration;"
            },
            "type": "LAMBDA",
            "decisionCases": {},
            "defaultCase": [],
            "forkTasks": [],
            "startDelay": 0,
            "joinOn": [],
            "optional": false,
            "taskDefinition": {
              "name": "func_1",
              "retryCount": 3,
              "timeoutSeconds": 10000,
              "inputKeys": [],
              "outputKeys": [],
              "timeoutPolicy": "TIME_OUT_WF",
              "retryLogic": "FIXED",
              "retryDelaySeconds": 0,
              "responseTimeoutSeconds": 3600,
              "inputTemplate": {},
              "rateLimitPerFrequency": 0,
              "rateLimitFrequencyInSeconds": 1
            },
            "defaultExclusiveJoinTask": [],
            "asyncComplete": false,
            "loopOver": []
          }
        ]
      }
    ],
    "inputParameters": [],
    "outputParameters": {},
    "schemaVersion": 2,
    "restartable": true,
    "workflowStatusListenerEnabled": true,
    "timeoutPolicy": "ALERT_ONLY",
    "timeoutSeconds": 0,
    "variables": {}
  },
  "active": true,
  "subWorkflow": false,
  "inputParameters": {},
  "draftDefinition": null,
  "draft": false,
  "version": 6,
  "id": "edeca117-8930-4bbf-9fc8-d24649156262",
  "createdBy": "test",
  "updatedDate": "2021-04-16T16:34:45+0000"
} 

Workflow execution in conductor:

{
  "name": "DoWhile Test",
  "version": 6,
  "priority": 0,
  "status": "RUNNING",
  "input": {},
  "workflowDefinition": {
    "updateTime": 1618590870149,
    "name": "DoWhile Test",
    "description": "do-while",
    "version": 6,
    "tasks": [
      {
        "name": "LoopTask",
        "taskReferenceName": "LoopTask",
        "inputParameters": {
          "value": 10
        },
        "type": "DOWHILE",
        "decisionCases": {},
        "defaultCase": [],
        "forkTasks": [],
        "startDelay": 0,
        "joinOn": [],
        "optional": false,
        "taskDefinition": {
          "name": "LoopTask",
          "retryCount": 3,
          "timeoutSeconds": 10000,
          "inputKeys": [],
          "outputKeys": [],
          "timeoutPolicy": "TIME_OUT_WF",
          "retryLogic": "FIXED",
          "retryDelaySeconds": 0,
          "responseTimeoutSeconds": 3600,
          "inputTemplate": {},
          "rateLimitPerFrequency": 0,
          "rateLimitFrequencyInSeconds": 1
        },
        "defaultExclusiveJoinTask": [],
        "asyncComplete": false,
        "loopCondition": "if ($.LoopTask['iteration'] < $.value) { true; } else { false; }",
        "loopOver": [
          {
            "name": "func_1",
            "taskReferenceName": "func_1",
            "inputParameters": {
              "scriptExpression": "$.LoopTask.output.iteration;"
            },
            "type": "LAMBDA",
            "decisionCases": {},
            "defaultCase": [],
            "forkTasks": [],
            "startDelay": 0,
            "joinOn": [],
            "optional": false,
            "taskDefinition": {
              "name": "func_1",
              "retryCount": 3,
              "timeoutSeconds": 10000,
              "inputKeys": [],
              "outputKeys": [],
              "timeoutPolicy": "TIME_OUT_WF",
              "retryLogic": "FIXED",
              "retryDelaySeconds": 0,
              "responseTimeoutSeconds": 3600,
              "inputTemplate": {},
              "rateLimitPerFrequency": 0,
              "rateLimitFrequencyInSeconds": 1
            },
            "defaultExclusiveJoinTask": [],
            "asyncComplete": false,
            "loopOver": []
          }
        ]
      }
    ],
    "inputParameters": [],
    "outputParameters": {},
    "schemaVersion": 2,
    "restartable": true,
    "workflowStatusListenerEnabled": true,
    "timeoutPolicy": "ALERT_ONLY",
    "timeoutSeconds": 0,
    "variables": {}
  },
  "taskToDomain": {},
  "terminalState": false,
  "state": "RUNNING",
  "ownerApp": "",
  "createTime": 1618590905662,
  "updateTime": 1618590907849,
  "endTime": 0,
  "workflowId": "e5b5f862-41f8-4522-bdd4-4bfb3ded8078",
  "tasks": [
    {
      "taskType": "DOWHILE",
      "status": "SCHEDULED",
      "inputData": {
        "value": 10
      },
      "referenceTaskName": "LoopTask",
      "retryCount": 0,
      "seq": 1,
      "pollCount": 0,
      "taskDefName": "LoopTask",
      "scheduledTime": 1618590907751,
      "startTime": 0,
      "endTime": 0,
      "updateTime": 0,
      "startDelayInSeconds": 0,
      "retried": false,
      "executed": false,
      "callbackFromWorker": true,
      "responseTimeoutSeconds": 0,
      "workflowInstanceId": "e5b5f862-41f8-4522-bdd4-4bfb3ded8078",
      "workflowType": "DoWhile Test",
      "taskId": "af940821-8701-4a9b-8e6a-2856c8ac0195",
      "callbackAfterSeconds": 0,
      "outputData": {},
      "workflowTask": {
        "name": "LoopTask",
        "taskReferenceName": "LoopTask",
        "inputParameters": {
          "value": 10
        },
        "type": "DOWHILE",
        "decisionCases": {},
        "defaultCase": [],
        "forkTasks": [],
        "startDelay": 0,
        "joinOn": [],
        "optional": false,
        "taskDefinition": {
          "name": "LoopTask",
          "retryCount": 3,
          "timeoutSeconds": 10000,
          "inputKeys": [],
          "outputKeys": [],
          "timeoutPolicy": "TIME_OUT_WF",
          "retryLogic": "FIXED",
          "retryDelaySeconds": 0,
          "responseTimeoutSeconds": 3600,
          "inputTemplate": {},
          "rateLimitPerFrequency": 0,
          "rateLimitFrequencyInSeconds": 1
        },
        "defaultExclusiveJoinTask": [],
        "asyncComplete": false,
        "loopCondition": "if ($.LoopTask['iteration'] < $.value) { true; } else { false; }",
        "loopOver": [
          {
            "name": "func_1",
            "taskReferenceName": "func_1",
            "inputParameters": {
              "scriptExpression": "$.LoopTask.output.iteration;"
            },
            "type": "LAMBDA",
            "decisionCases": {},
            "defaultCase": [],
            "forkTasks": [],
            "startDelay": 0,
            "joinOn": [],
            "optional": false,
            "taskDefinition": {
              "name": "func_1",
              "retryCount": 3,
              "timeoutSeconds": 10000,
              "inputKeys": [],
              "outputKeys": [],
              "timeoutPolicy": "TIME_OUT_WF",
              "retryLogic": "FIXED",
              "retryDelaySeconds": 0,
              "responseTimeoutSeconds": 3600,
              "inputTemplate": {},
              "rateLimitPerFrequency": 0,
              "rateLimitFrequencyInSeconds": 1
            },
            "defaultExclusiveJoinTask": [],
            "asyncComplete": false,
            "loopOver": []
          }
        ]
      },
      "rateLimitPerFrequency": 0,
      "rateLimitFrequencyInSeconds": 1,
      "workflowPriority": 0,
      "iteration": 0,
      "queueWaitTime": 0,
      "taskDefinition": {
        "present": true
      },
      "loopOverTask": false,
      "taskStatus": "SCHEDULED"
    }
  ],
  "output": {},
  "workflowType": "DoWhile Test",
  "schemaVersion": 2,
  "failedReferenceTaskNames": [],
  "variables": {},
  "lastRetriedTime": 0,
  "workflowName": "DoWhile Test",
  "workflowVersion": 6,
  "startTime": 1618590905662
}

As you can see in above workflow execution, that do-while task is in SCHEDULED state. My use case here is to execute the lambda function and HTTP task as part of the do-while loop task. For now, I am just returning the iteration value from lambda task, which in turn I would like to use for my HTTP task to execute some http api as long as the loop returns true.

Could you please point out what am I doing wrong or missing here?

Thanks in advance.
Ketan

[maheshyaddanapudi]

Hi Ketan,

There are a couple of things which I think might be the reason.

1. The $.LoopTask is not accessible inside the loop where the LAMBDA task is defined. Even if the was executed i.e. went beyond SCHEDULED, it would fail and LAMBDA task status will be set as CANCELLED.

2. The task definition for a System Task should not be included in the workflow definition request. This at times, per my observation, makes conductor not pick the System Task up and is left in SCHEDULED status forever before the Workflow TIMED_OUT.

Try the below sample workflow definition.

{
    "ownerApp": "Test",
    "createdBy": "Test",
    "name": "TEST_DO_WHILE",
    "description": "Testing Do While",
    "version": 1,
    "tasks": [
      {
        "name": "Loop Task",
        "taskReferenceName": "LoopTask",
        "type": "DO_WHILE",
        "inputParameters": {
          "value": 10
        },
        "optional": false,
        "asyncComplete": false,
        "loopCondition": "if ($.LoopTask['iteration'] < $.value ) { true; } else { false; }",
        "loopOver": [
          {
              "name": "first task",
              "taskReferenceName": "first_task",
              "inputParameters": {
                  "http_request": {
                      "uri": "${workflow.input.url}",
                      "method": "${workflow.input.method}"
                  }
              },
              "type": "HTTP"
          },
          {
            "name": "second task",
            "taskReferenceName": "second_task",
            "type": "LAMBDA",
            "inputParameters": {
                "lambdaValue": "${workflow.input.lambdaValue}",
                "scriptExpression": "if ($.lambdaValue > 0 ) { return {testvalue: true} } else { return {testvalue: false} }"
            }
          }
        ]
      }
    ],
    "inputParameters":["url", "method", "lambdaValue"],
    "outputParameters": {
      "do_while_output": "${LoopTask.output}"
    },
    "schemaVersion": 2,
    "restartable": true,
    "workflowStatusListenerEnabled": false,
    "ownerEmail": "abc@xyz.com",
    "timeoutPolicy": "ALERT_ONLY",
    "timeoutSeconds": 30
  }

This workflow contains a DO_WHILE loop which iterates 10 times. In each iteration it would perform two tasks. For the first task, it would call an API i.e. HTTP task and for the second, it would perform a LAMBDA task. For both the tasks the inputs are taken from Workflow input and the output of the entire DO_WHILE task is set as the output of the workflow itself.

To execute the workflow, try the below request.

 {
   "name": "TEST_DO_WHILE",
   "version": 1,
   "correlationId": "WF_TEST_DO_WHILE",
   "input": {
     "lambdaValue": 10,
     "url": "http://localhost:8080/api/health",
     "method": "GET"
     }
 }

The workflow should ideally get executed and the output of DO_WHILE task would be set as Workflow output.

Thanks and Regards,
Mahesh Yaddanapudi.

[ketantalreja]

Hi Mahesh,

Thanks a lot for your inputs, I tried out without having the task definition inside workflow definition and it worked 👍

One thing I fail to understand is I have been using other System task such as DECISION, LAMBDA, HTTP, etc. and all of them works with task definition inside workflow definition, but with the exception type DOWHILE system task.

@apanicker-nflx could you provide more details on this behavior with DOWHILE system task?

Thanks,
Ketan

[maheshyaddanapudi]

Hi Mahesh,

Thanks a lot for your inputs, I tried out without having the task definition inside workflow definition and it worked 👍

One thing I fail to understand is I have been using other System task such as DECISION, LAMBDA, HTTP, etc. and all of them works with task definition inside workflow definition, but with the exception type DOWHILE system task.

@apanicker-nflx could you provide more details on this behavior with DOWHILE system task?

Thanks,
Ketan

Hi Ketan,

From what I can understand by seeing the code at a high level, the only use of task definition for a system task, is if it has an input template. I couldn't go through full code as of now. I will update if I get some clues in the correlation between task definition and system task.

[apanicker-nflx]

Task definitions can be provided as part of system task configuration within a workflow definition.
Apart from input templates, task definitions can be used to provide configurations like retry counts, timeout etc.

have been using other System task such as DECISION, LAMBDA, HTTP, etc. and all of them works with task definition inside workflow definition, but with the exception type DOWHILE system task.

In the case of DO_WHILE task, this should also be the case. We would need to investigate this behavior and make sure that DO_WHILE also adheres to the same contract as other system tasks.