如何在AWS CDK中获取StepFunctionsStartExecution的返回结果？

业务场景 #

• 在Step Function 1中遍历子节点
• 将每个JSON对象发送至Step Function 2
• Step Function 2执行结束后接收第三方API的响应
• 在Step Function 1中使用该API响应

当前代码示例 #

this.mapOverChildren = new Map(this, "Map Over Children", {
  itemsPath: "$.data.children",
  maxConcurrency: 1,
  resultPath: "$.children",
});

this.mapOverChildren.iterator(new StepFunctionsStartExecution(this, 'Send Data to Step Function 2', {
  stateMachine: stepFunction.Two,
  integrationPattern: IntegrationPattern.REQUEST_RESPONSE,
  resultPath: "$.stepFunctionTwoResponse"
});

当前问题 #

使用上述配置时，resultPath返回的仅为Step Function 2的执行元数据，而非其实际输出内容：

{
    "ExecutionArn": "xxxxxxxxxxxx",
    "SdkHttpMetadata": {
      "AllHttpHeaders": {
        "x-amzn-RequestId": [
          "xxxxxxxxxxx"
        ],
        "Content-Length": [
          "000"
        ],
        "Date": [
          "Thu, 23 Feb 2023 11:24:14 GMT"
        ],
        "Content-Type": [
          "application/xxxxxxxxxxxxxxxxxxxxxxx"
        ]
      },
      "HttpHeaders": {
        "Content-Length": "111",
        "Content-Type": "application/xxxxxxxxxxxxxxxxxxxxxxx",
        "Date": "Thu, 23 Feb 2023 11:24:14 GMT",
        "x-amzn-RequestId": "xxxxxxxxxxxxxxxxxxxxxxx"
      },
      "HttpStatusCode": 200
    },
    "SdkResponseMetadata": {
      "RequestId": "xxxxxxxxxxxxxxxxxxxxxxx"
    },
    "StartDate": 1677151454840
  }

解决方案 #

要获取Step Function 2的返回数据，需要调整集成模式并修改子状态机的结尾逻辑：

1. 修改父状态机的集成配置 #

将IntegrationPattern.REQUEST_RESPONSE替换为IntegrationPattern.WAIT_FOR_TASK_TOKEN，并将任务令牌传递给子状态机：

this.mapOverChildren.iterator(new StepFunctionsStartExecution(this, 'Send Data to Step Function 2', {
  stateMachine: stepFunction.Two,
  integrationPattern: IntegrationPattern.WAIT_FOR_TASK_TOKEN,
  // 将原始输入和任务令牌一起传递给子状态机
  input: JsonPath.stringAt('$'),
  taskTokenPath: '$.taskToken', // 指定任务令牌在输入中的存储路径
  resultPath: "$.stepFunctionTwoResponse"
}));

2. 调整子状态机（Step Function 2）的结尾 #

在Step Function 2的最后一步添加SendTaskSuccess状态，将第三方API的响应作为结果发送回父状态机：

// 假设子状态机中调用第三方API的Lambda节点
const callThirdPartyApi = new LambdaInvoke(this, 'Call Third Party API', {
  lambdaFunction: thirdPartyApiLambda,
  resultPath: "$.apiResponse"
});

// 添加SendTaskSuccess节点，将API响应返回给父状态机
const sendTaskSuccess = new SendTaskSuccess(this, 'Send Task Success to Parent', {
  taskToken: JsonPath.stringAt('$.taskToken'),
  output: JsonPath.stringAt('$.apiResponse') // 传递第三方API的响应数据
});

// 构建子状态机流程
callThirdPartyApi.next(sendTaskSuccess);

说明 #

• WAIT_FOR_TASK_TOKEN模式会让父状态机暂停执行，直到子状态机通过SendTaskSuccess返回结果
• 子状态机必须接收并保存父状态机传递的taskToken，才能完成结果回传
• 最终父状态机的resultPath将存储子状态机返回的第三方API响应数据

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