👋

【AWS CDK】JSONataなDistributedMapの中に書くアクションはJSONata形式で定義しよう【Sfn】

に公開
AWS
AWS Lambda
Step Functions
AWS CDK
tech

TL;DR


// Lambda
const functionState = sfntasks.LambdaInvoke.jsonata(this, 'Function', {
  lambdaFunction: props.function,
  taskTimeout: sfn.Timeout.duration(cdk.Duration.minutes(15)),
});

// DistributedMap
const distributedMap = sfn.DistributedMap.jsonata(this, 'Map', {
  maxConcurrency: 100,
  items: sfn.ProvideItems.jsonata('{% $states.input.Payload.items %}'),
});

distributedMap.itemProcessor(functionState);

経緯

Step FunctionsをCDKで記載する際に以下のように定義していました。

const stateMachine = new sfn.StateMachine(this, 'StateMachine', {
  comment: `Workflow`,
  logs: {
    destination: logGroup,
    includeExecutionData: true,
    level: sfn.LogLevel.ALL,
  },
  definitionBody: sfn.DefinitionBody.fromChainable(flowDefinition.chain),
  queryLanguage: sfn.QueryLanguage.JSONATA,
});

queryLanguage: sfn.QueryLanguage.JSONATA, を定義しているため、デフォルトで全てのアクションはJsonata形式になる想定でした。

しかし、以下のように記載した場合エラーとなりました。


// Lambda
const functionState = new sfntasks.LambdaInvoke(this, 'Function', {
  lambdaFunction: props.function,
  taskTimeout: sfn.Timeout.duration(cdk.Duration.minutes(15)),
});

// DistributedMap
const distributedMap = new sfn.DistributedMap(this, 'Map', {
  maxConcurrency: 100,
  items: sfn.ProvideItems.jsonata('{% $states.input.Payload.items %}'),
});

distributedMap.itemProcessor(functionState);

エラー内容は

❌  SampleCdk failed: ToolkitError: The stack named SampleCdk failed to deploy: UPDATE_ROLLBACK_COMPLETE: Resource handler returned message: "Invalid State Machine Definition: 'SCHEMA_VALIDATION_FAILED: The QueryLanguage is set to 'JSONata', but field 'Parameters' is only supported for the 'JSONPath' QueryLanguage at /States/distributedMap/ItemProcessor/States/Function, SCHEMA_VALIDATION_FAILED: Arguments field is required for resource ARN: arn:aws:states:::lambda:invoke at /States/distributedMap/ItemProcessor/States/Function/Resource' (Service: Sfn, Status Code: 400, Request ID: xxx) (SDK Attempt Count: 1)" (RequestToken: xxx, HandlerErrorCode: InvalidRequest)

簡単に言うと、
「DistributedMap を JSONata モードで使っているのに、LambdaInvoke が JSONPath のフィールドを勝手に生成してしまった」

というもの、の様です。
根本原因は、LambdaInvokeがJSONPathで作成されていることです。

DistributedMap外で作成されたLambdaは、sfn.StateMachineでJSONataモードで作成されますが、DistributedMap内のデフォルトは、JSONPathらしいです。

ですので、以下のように明示的にJSONataを宣言する必要があります。

const functionState = sfntasks.LambdaInvoke.jsonata(this, 'Function', {
  lambdaFunction: props.function,
  taskTimeout: sfn.Timeout.duration(cdk.Duration.minutes(15)),
});

おわりに

DistributedMap外と内で、queryLanguage: sfn.QueryLanguage.JSONATA,は引き継がれないんですね！
このエラーが出てしまった人は、JSONataなのか、JSONPathなのか確認してみてください！

Discussion