AWS Step Functions并行执行中Error_Handler如何访问st_time变量

问题分析 #

你在Parallel分支的Task_A中通过Assign定义的st_time是分支内部的本地变量，默认不会传递到Parallel状态的Catch流程（Error_Handler）中，所以直接访问无法获取到该值。需要把变量提升到Parallel状态的全局上下文，才能被Error_Handler访问。

解决方案 #

步骤1：将st_time存储到Parallel状态的上下文 #

修改Task_A中的Assign语法，使用$$引用父级（Parallel_Execution）的上下文，把st_time赋值到全局上下文：

"Task_A": {
  "Comment": "Invoke Lambda Function A to update db",
  "Type": "Task",
  "Resource": "arn:aws:lambda:region:account-id:function:FunctionA",
  "Assign": {
    "$$.st_time": "{% $states.result.st_time %}"
  },
  "Next": "Task_B"
}

这里$$代表当前状态的父级上下文，也就是Parallel_Execution的上下文，这样st_time就会被存在全局范围内，而非分支内部。

步骤2：在Error_Handler中访问st_time #

在Error_Handler中，通过JSONata的$states.context来获取全局上下文里的st_time。可以直接在Input中传递给Lambda：

"Error_Handler": {
  "Comment": "Update db to reflect error in above process",
  "Type": "Task",
  "Resource": "arn:aws:lambda:region:account-id:function:ErrorHandlerFunction",
  "Input": {
    "st_time": "{% $states.context.st_time %}",
    "error_details": "{% $states.error %}"
  },
  "Next": "Final_State"
}

这样你的ErrorHandlerFunction就能接收到st_time参数，用于后续的数据库更新操作。

修改后的完整状态机代码 #

{
  "StartAt": "Assign Variables",
  "QueryLanguage": "JSONata",
  "States": {
    "Assign Variables": {
      "Comment": "Assign variables from input",
      "Type": "Pass",
      "Assign": {
        "var1": "{% $states.input.var1 %}",
        "var2": "{% $states.input.var2 %}"
      },
      "Next": "Parallel_Execution"
    },
    "Parallel_Execution": {
      "Type": "Parallel",
      "Branches": [
        {
          "StartAt": "Task_A",
          "States": {
            "Task_A": {
              "Comment": "Invoke Lambda Function A to update db",
              "Type": "Task",
              "Resource": "arn:aws:lambda:region:account-id:function:FunctionA",
              "Assign": {
                "$$.st_time": "{% $states.result.st_time %}"
              },
              "Next": "Task_B"
            },
            "Task_B": {
              "Comment": "Invoke Lambda Function B to send notification",
              "Type": "Task",
              "Resource": "arn:aws:lambda:region:account-id:function:FunctionB",
              "End": true
            }
          }
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["States.ALL"],
          "Next": "Error_Handler"
        }
      ],
      "End": true
    },
    "Error_Handler": {
      "Comment": "Update db to reflect error in above process",
      "Type": "Task",
      "Resource": "arn:aws:lambda:region:account-id:function:ErrorHandlerFunction",
      "Input": {
        "st_time": "{% $states.context.st_time %}",
        "error_details": "{% $states.error %}"
      },
      "Next": "Final_State"
    },
    "Final_State": {
      "Type": "Fail"
    }
  }
}

内容的提问来源于stack exchange，提问作者Suhas Shravagi