Conductor: Exclusive Join not working when embedded in Fork/Join

[RemcoBuddelmeijer]

Exclusive Join not working when embedded in Fork/Join

Problem

Hello,

In one of our workflows we have a very specific with the following (generic) flow:
Note: the task types assigned to any TASK step(s) are not bound to any specific task type, this can be either a JOIN/FORK, LAMBDA, SIMPLE, etc.. task type.

FORK(T1) -> |
             TASK*(T2)                                          -> |
             DECISION(T2) -> TASK*(T3|T4) -> EXCLUSIVE JOIN(T5) -> |
                                                                    JOIN(T6)

When we embed a decision with an exclusive join inside another fork/join, the workflow will not continue processing after the exclusive join and with that never completing the parent join.
This only happens when we have the parent join (T6), join on the exclusive join (T5) (which is valid, see "Working alternative").

Failing decision metadata representation:

{
  "name" : "exclusive_join_test",
  "description" : "Exclusive Join Test",
  "version" : 1,
  "tasks" : [
    {
      "taskReferenceName" : "fork",
      "name" : "fork",
      "forkTasks" : [
        [
          {
            "name" : "lambda_nothing",
            "taskReferenceName" : "lambda_nothing",
            "inputParameters" : {
              "scriptExpression" : "return {value: \"nothing\"}"
            },
            "type" : "LAMBDA",
            "optional" : false,
            "asyncComplete" : false
          }
        ],
        [
          {
            "name" : "decision_test",
            "taskReferenceName" : "decision_test",
            "inputParameters" : {
              "case_value_param" : "${workflow.input.decision}"
            },
            "type" : "DECISION",
            "caseValueParam" : "case_value_param",
            "decisionCases" : {
              "true" : [
                {
                  "taskReferenceName" : "lambda_true",
                  "name" : "lambda_true",
                  "inputParameters" : {
                    "value" : "${workflow.input.decision}",
                    "scriptExpression" : "return {value: $.value}"
                  },
                  "optional" : false,
                  "type" : "LAMBDA",
                  "asyncComplete" : false
                }
              ]
            },
            "defaultCase" : [
              {
                "name" : "lambda_false",
                "taskReferenceName" : "lambda_false",
                "inputParameters" : {
                  "value" : "${workflow.input.decision}",
                  "scriptExpression" : "return {value: $.value}"
                },
                "type" : "LAMBDA",
                "optional" : false,
                "asyncComplete" : false
              }
            ],
            "optional" : false,
            "asyncComplete" : false
          },
          {
            "name" : "exclusive_join",
            "taskReferenceName" : "exclusive_join",
            "type" : "EXCLUSIVE_JOIN",
            "joinOn" : [
              "lambda_true",
              "lambda_false"
            ],
            "optional" : false,
            "defaultExclusiveJoinTask" : [
              "decision_test"
            ],
            "asyncComplete" : false
          }
        ]
      ],
      "type" : "FORK_JOIN"
    },
    {
      "name" : "join",
      "taskReferenceName" : "join",
      "joinOn" : [
        "lambda_nothing",
        "exclusive_join"
      ],
      "type" : "JOIN"
    }
  ],
  "inputParameters" : [
    "decision"
  ],
  "schemaVersion" : 2,
  "restartable" : true,
  "workflowStatusListenerEnabled" : false
}

Working alternative

However, when we isolate the same exclusive join outside of the fork/join, the exclusive join completes and the workflow continues as expected:

Working decision metadata representation:

{
  "name" : "exclusive_join_test",
  "description" : "Exclusive Join Test",
  "version" : 1,
  "tasks" : [
    {
      "name" : "decision_test",
      "taskReferenceName" : "decision_test",
      "inputParameters" : {
        "case_value_param" : "${workflow.input.decision}"
      },
      "type" : "DECISION",
      "caseValueParam" : "case_value_param",
      "decisionCases" : {
        "true" : [
          {
            "taskReferenceName" : "lambda_true",
            "name" : "lambda_true",
            "inputParameters" : {
              "value" : "${workflow.input.decision}",
              "scriptExpression" : "return {value: $.value}"
            },
            "optional" : false,
            "type" : "LAMBDA",
            "asyncComplete" : false
          }
        ]
      },
      "defaultCase" : [
        {
          "name" : "lambda_false",
          "taskReferenceName" : "lambda_false",
          "inputParameters" : {
            "value" : "${workflow.input.decision}",
            "scriptExpression" : "return {value: $.value}"
          },
          "type" : "LAMBDA",
          "optional" : false,
          "asyncComplete" : false
        }
      ],
      "optional" : false,
      "asyncComplete" : false
    },
    {
      "name" : "exclusive_join",
      "taskReferenceName" : "exclusive_join",
      "type" : "EXCLUSIVE_JOIN",
      "joinOn" : [
        "lambda_true",
        "lambda_false"
      ],
      "optional" : false,
      "defaultExclusiveJoinTask" : [
        "decision_test"
      ],
      "asyncComplete" : false
    }
  ],
  "inputParameters" : [
    "decision"
  ],
  "schemaVersion" : 2,
  "restartable" : true,
  "workflowStatusListenerEnabled" : false
}

Some metadata:

Conductor version: 2.10.2

[jkaipa]

Looks like this is linked/similar to https://github.com/Netflix/conductor/issues/644
Also refer to https://github.com/Netflix/conductor/issues/881#issuecomment-488825067

[RemcoBuddelmeijer]

I looked at both the issues, I believe they are indeed connected in a certain way.

After looking at both #644 and #881 I went to look for a workaround.
First, to validate the connection between both tickets, I changed my workflow to have yet another decision in the FORK/JOIN. Now the other task in the FORK is no longer continuing after the decision, so that bug is connected to here (see image 1).

Secondly and however, I did find a way on how to have the same expected output, but in a different execution order (put the exclusive join after the parent join) (similar to "fix" in #881), for that: see image 2.
I hope that this is of any help with understanding what might go wrong, even though this is still not the desired fix/solution.

No logs were found that might lead to a resolution. All logs that were given are for the LAMBDA tasks.

The content of the workflow definition:

{
  "name" : "exclusive_join_test",
  "description" : "Exclusive Join Test",
  "version" : 1,
  "tasks" : [
    {
      "taskReferenceName" : "fork",
      "name" : "fork",
      "forkTasks" : [
        [
          {
            "name" : "lambda_fork",
            "taskReferenceName" : "lambda_fork",
            "forkTasks" : [
              [
                {
                  "taskReferenceName" : "lambda_something",
                  "name" : "lambda_something",
                  "inputParameters" : {
                    "scriptExpression" : "return {value: \"something\"}"
                  },
                  "optional" : false,
                  "type" : "LAMBDA",
                  "asyncComplete" : false
                }
              ],
              [
                {
                  "taskReferenceName" : "lambda_nothing",
                  "name" : "lambda_nothing",
                  "inputParameters" : {
                    "scriptExpression" : "return {value: \"nothing\"}"
                  },
                  "optional" : false,
                  "type" : "LAMBDA",
                  "asyncComplete" : false
                }
              ]
            ],
            "type" : "FORK_JOIN",
            "optional" : false,
            "asyncComplete" : false
          },
          {
            "name" : "lambda_join",
            "taskReferenceName" : "lambda_join",
            "joinOn" : [
              "lambda_nothing",
              "lambda_something"
            ],
            "type" : "JOIN"
          },
          {
            "name" : "decision_lambda",
            "taskReferenceName" : "decision_lambda",
            "inputParameters" : {
              "case_value_param" : "${lambda_join.output.lambda_something.result.value}"
            },
            "type" : "DECISION",
            "caseValueParam" : "case_value_param",
            "decisionCases" : {
              "something" : [
                {
                  "taskReferenceName" : "lambda_otherthing",
                  "name" : "lambda_otherthing",
                  "inputParameters" : {
                    "scriptExpression" : "return {value: \"otherthing\"}"
                  },
                  "optional" : false,
                  "type" : "LAMBDA",
                  "asyncComplete" : false
                }
              ]
            },
            "defaultCase" : [
              {
                "name" : "lambda_mything",
                "taskReferenceName" : "lambda_mything",
                "inputParameters" : {
                  "scriptExpression" : "return {value: \"mything\"}"
                },
                "type" : "LAMBDA",
                "optional" : false,
                "asyncComplete" : false
              }
            ],
            "optional" : false,
            "asyncComplete" : false
          },
          {
            "name" : "lambda_overload",
            "taskReferenceName" : "lambda_overload",
            "inputParameters" : {
              "scriptExpression" : "return {value: \"overload\"}"
            },
            "type" : "LAMBDA",
            "optional" : false,
            "asyncComplete" : false
          }
        ],
        [
          {
            "name" : "decision_test",
            "taskReferenceName" : "decision_test",
            "inputParameters" : {
              "case_value_param" : "${workflow.input.decision}"
            },
            "type" : "DECISION",
            "caseValueParam" : "case_value_param",
            "decisionCases" : {
              "true" : [
                {
                  "taskReferenceName" : "lambda_true",
                  "name" : "lambda_true",
                  "inputParameters" : {
                    "value" : "${workflow.input.decision}",
                    "scriptExpression" : "return {value: $.value}"
                  },
                  "optional" : false,
                  "type" : "LAMBDA",
                  "asyncComplete" : false
                }
              ]
            },
            "defaultCase" : [
              {
                "name" : "lambda_false",
                "taskReferenceName" : "lambda_false",
                "inputParameters" : {
                  "value" : "${workflow.input.decision}",
                  "scriptExpression" : "return {value: $.value}"
                },
                "type" : "LAMBDA",
                "optional" : false,
                "asyncComplete" : false
              }
            ],
            "optional" : false,
            "asyncComplete" : false
          },
          {
            "name" : "exclusive_join",
            "taskReferenceName" : "exclusive_join",
            "type" : "EXCLUSIVE_JOIN",
            "joinOn" : [
              "lambda_true",
              "lambda_false"
            ],
            "optional" : false,
            "defaultExclusiveJoinTask" : [
              "decision_test"
            ],
            "asyncComplete" : false
          }
        ]
      ],
      "type" : "FORK_JOIN"
    },
    {
      "name" : "join",
      "taskReferenceName" : "join",
      "joinOn" : [
        "lambda_overload",
        "exclusive_join"
      ],
      "type" : "JOIN"
    }
  ],
  "inputParameters" : [
    "decision"
  ],
  "schemaVersion" : 2,
  "restartable" : true,
  "workflowStatusListenerEnabled" : false
}

  "name" : "exclusive_join_test",
  "description" : "Exclusive Join Test",
  "version" : 1,
  "tasks" : [
    {
      "taskReferenceName" : "fork",
      "name" : "fork",
      "forkTasks" : [
        [
          {
            "name" : "lambda_fork",
            "taskReferenceName" : "lambda_fork",
            "forkTasks" : [
              [
                {
                  "taskReferenceName" : "lambda_something",
                  "name" : "lambda_something",
                  "inputParameters" : {
                    "scriptExpression" : "return {value: \"something\"}"
                  },
                  "optional" : false,
                  "type" : "LAMBDA",
                  "asyncComplete" : false
                }
              ],
              [
                {
                  "taskReferenceName" : "lambda_nothing",
                  "name" : "lambda_nothing",
                  "inputParameters" : {
                    "scriptExpression" : "return {value: \"nothing\"}"
                  },
                  "optional" : false,
                  "type" : "LAMBDA",
                  "asyncComplete" : false
                }
              ]
            ],
            "type" : "FORK_JOIN",
            "optional" : false,
            "asyncComplete" : false
          },
          {
            "name" : "lambda_join",
            "taskReferenceName" : "lambda_join",
            "joinOn" : [
              "lambda_nothing",
              "lambda_something"
            ],
            "type" : "JOIN"
          },
          {
            "name" : "decision_lambda",
            "taskReferenceName" : "decision_lambda",
            "inputParameters" : {
              "case_value_param" : "${lambda_join.output.lambda_something.result.value}"
            },
            "type" : "DECISION",
            "caseValueParam" : "case_value_param",
            "decisionCases" : {
              "something" : [
                {
                  "taskReferenceName" : "lambda_otherthing",
                  "name" : "lambda_otherthing",
                  "inputParameters" : {
                    "scriptExpression" : "return {value: \"otherthing\"}"
                  },
                  "optional" : false,
                  "type" : "LAMBDA",
                  "asyncComplete" : false
                }
              ]
            },
            "defaultCase" : [
              {
                "name" : "lambda_mything",
                "taskReferenceName" : "lambda_mything",
                "inputParameters" : {
                  "scriptExpression" : "return {value: \"mything\"}"
                },
                "type" : "LAMBDA",
                "optional" : false,
                "asyncComplete" : false
              }
            ],
            "optional" : false,
            "asyncComplete" : false
          }
        ],
        [
          {
            "name" : "decision_test",
            "taskReferenceName" : "decision_test",
            "inputParameters" : {
              "case_value_param" : "${workflow.input.decision}"
            },
            "type" : "DECISION",
            "caseValueParam" : "case_value_param",
            "decisionCases" : {
              "true" : [
                {
                  "taskReferenceName" : "lambda_true",
                  "name" : "lambda_true",
                  "inputParameters" : {
                    "value" : "${workflow.input.decision}",
                    "scriptExpression" : "return {value: $.value}"
                  },
                  "optional" : false,
                  "type" : "LAMBDA",
                  "asyncComplete" : false
                }
              ]
            },
            "defaultCase" : [
              {
                "name" : "lambda_false",
                "taskReferenceName" : "lambda_false",
                "inputParameters" : {
                  "value" : "${workflow.input.decision}",
                  "scriptExpression" : "return {value: $.value}"
                },
                "type" : "LAMBDA",
                "optional" : false,
                "asyncComplete" : false
              }
            ],
            "optional" : false,
            "asyncComplete" : false
          }
        ]
      ],
      "type" : "FORK_JOIN"
    },
    {
      "name" : "join",
      "taskReferenceName" : "join",
      "joinOn" : [
        "decision_lambda"
      ],
      "type" : "JOIN"
    },
    {
      "name" : "exclusive_join",
      "taskReferenceName" : "exclusive_join",
      "type" : "EXCLUSIVE_JOIN",
      "joinOn" : [
        "lambda_true",
        "lambda_false"
      ],
      "optional" : false,
      "asyncComplete" : false
    }
  ],
  "inputParameters" : [
    "decision"
  ],
  "schemaVersion" : 2,
  "restartable" : true,
  "workflowStatusListenerEnabled" : false
}

[apanicker-nflx]

Thanks for reporting the issue. We will look into fixing this issue soon. However, feel free to submit a PR with the fix if you need this sooner.

[jkaipa]

@apanicker-nflx This is a duplicate of https://github.com/Netflix/conductor/issues/644 .

[apanicker-nflx]

@jkaipa Yes, this seems to be an edge-case issue related to how the task order is determined.

[apanicker-nflx]

Fixed as part of #1449