aws-stepfunctions: Distributed Map State Item Processor executionType doesn't recognize ProcessorType.EXPRESS

[Commas929]

Describe the bug

I have the following piece of CDK code but when I deploy it and check in AWS console, the step function execution type for the map run is Standard instead of Express.

The code:

    const qotOverrideDistributedMapState = new DistributedMap(this, 'Distributed Map State', {
      maxConcurrency: 1000,
      itemBatcher: new ItemBatcher({
        maxItemsPerBatch: 10
      }),
      itemReader: new S3CsvItemReader({
        bucket: qotOverrideBucket,
        key: JsonPath.stringAt('$.s3Key'),
        csvHeaders: CsvHeaders.useFirstRow()
      }),
      resultWriter: new ResultWriter({
        bucket: qotOverrideBucket,
        prefix: 'QoTOverrideProcessingResult',
      }),
      toleratedFailureCount: 0,
    }).itemProcessor(new LambdaInvoke(this, 'Invoke Lambda Processor', {
      lambdaFunction: qotOverrideProcessor,
    }), {
      mode: ProcessorMode.DISTRIBUTED,
      executionType: ProcessorType.EXPRESS,
    })

    const qotOverrideFileProcessor = new StateMachine(this, 'QoT Override File Processor', {
      role: Role.fromRoleArn(this, 'QoT-Override-File-Processor-SfnRole', qotOverrideSfnRole.roleArn, {
        mutable: false,
      }),
      definitionBody: DefinitionBody.fromChainable(qotOverrideDistributedMapState),
      stateMachineName: 'QoT-Override-File-Processor-Sfn',
      stateMachineType: StateMachineType.STANDARD,
      logs: {
        destination: new LogGroup(this, 'QoT-Override-File-Processor-LogGroup', {
          retention: RetentionDays.THREE_MONTHS,
          logGroupName: 'QoT-Override-File-Processor-Log',
        }),
        level: LogLevel.ERROR,
        includeExecutionData: true,
      },
    })

The generated State Machine definition copied from AWS console:

{
  "StartAt": "Distributed Map State",
  "States": {
    "Distributed Map State": {
      "Type": "Map",
      "End": true,
      "ItemProcessor": {
        "ProcessorConfig": {
          "Mode": "DISTRIBUTED",
          "ExecutionType": "STANDARD"
        },
        "StartAt": "Invoke Lambda Processor",
        "States": {
          "Invoke Lambda Processor": {
            "End": true,
            "Retry": [
              {
                "ErrorEquals": [
                  "Lambda.ClientExecutionTimeoutException",
                  "Lambda.ServiceException",
                  "Lambda.AWSLambdaException",
                  "Lambda.SdkClientException"
                ],
                "IntervalSeconds": 2,
                "MaxAttempts": 6,
                "BackoffRate": 2
              }
            ],
            "Type": "Task",
            "Resource": "arn:aws:states:::lambda:invoke",
            "Parameters": {
              "FunctionName": "arn:aws:lambda:us-east-1:502523373620:function:QoTOverrideProcessor",
              "Payload.$": "$"
            }
          }
        }
      },
      "MaxConcurrency": 1000,
      "ItemReader": {
        "Resource": "arn:aws:states:::s3:getObject",
        "ReaderConfig": {
          "InputType": "CSV",
          "CSVHeaderLocation": "FIRST_ROW"
        },
        "Parameters": {
          "Bucket.$": "$.s3Bucket",
          "Key.$": "$.s3Key"
        }
      },
      "ItemBatcher": {
        "MaxItemsPerBatch": 10
      },
      "ResultWriter": {
        "Resource": "arn:aws:states:::s3:putObject",
        "Parameters": {
          "Bucket.$": "$.s3Bucket",
          "Prefix": "QoTOverrideProcessingResult"
        }
      }
    }
  }
}

Expected Behavior

The ItemProcessor execution type should be Express

Current Behavior

The ItemProcessor execution type is Standard and the specified type in CDK is ignored.

Reproduction Steps

    const bucket = new Bucket(this, 'Bucket', {});
    const distributedMapState = new DistributedMap(this, 'Distributed Map State', {
      maxConcurrency: 1000,
      itemReader: new S3CsvItemReader({
        bucket: bucket,
        key: JsonPath.stringAt('$.s3Key'),
        csvHeaders: CsvHeaders.useFirstRow()
      }),
      toleratedFailureCount: 0,
    }).itemProcessor(new Pass(this, 'Pass State', {}), {
      mode: ProcessorMode.DISTRIBUTED,
      executionType: ProcessorType.EXPRESS,
    })

    const fileProcessor = new StateMachine(this, 'File Processor', {
      definitionBody: DefinitionBody.fromChainable(distributedMapState),
      stateMachineName: 'File-Processor-Sfn',
      stateMachineType: StateMachineType.STANDARD,
    })

Possible Solution

No response

Additional Information/Context

No response

CDK CLI Version

2.136.0

Framework Version

No response

Node.js Version

NodeJS-18

OS

macOS Sonoma 14.4.1

Language

TypeScript

Language Version

No response

Other information

No response

[khushail]

Hey @Commas929 , thanks for reporting this bug.

On investigating this issue, I checked the CDK Docs and found that this PR introduced the support for itemProcessor() where executiontype can be set to required enum value-

aws-cdk/packages/aws-cdk-lib/aws-stepfunctions/lib/states/distributed-map.ts

Line 225 in d7e6a10

public itemProcessor(processor: IChainable, config: ProcessorConfig = {}): DistributedMap {

However later on this value was overwritten by change introduced in this PR -

aws-cdk/packages/aws-cdk-lib/aws-stepfunctions/lib/states/distributed-map.ts

Line 238 in d7e6a10

rendered.ItemProcessor.ProcessorConfig.ExecutionType = this.mapExecutionType;

So I tried reunning the code with giving value of mapExecutionType to EXPRESS and it worked by assigning the same value to executionType. However I feel this might not be intended as such.

Sharing the code -

    const distributedMapState = new DistributedMap(this, 'Distributed Map State', {
      maxConcurrency: 1000,
      itemReader: new S3CsvItemReader({
        bucket: bucket,
        key: JsonPath.stringAt('$.s3Key'),
        csvHeaders: CsvHeaders.useFirstRow()
      }),
      mapExecutionType: StateMachineType.EXPRESS,
      resultPath: JsonPath.DISCARD,
      toleratedFailureCount: 0,
    }).itemProcessor(new Pass(this, 'Pass State', {}), {
      mode: ProcessorMode.DISTRIBUTED,
      executionType: ProcessorType.STANDARD,
    })

    const fileProcessor = new StateMachine(this, 'File Processor', {
      definitionBody: DefinitionBody.fromChainable(distributedMapState),
      stateMachineName: 'File-Processor-Sfn',
      stateMachineType: StateMachineType.STANDARD,
    })

and synthesized template -

"FileProcessor4197797F": {
   "Type": "AWS::StepFunctions::StateMachine",
   "Properties": {
    "DefinitionString": {
     "Fn::Join": [
      "",
      [
       "{\"StartAt\":\"Distributed Map State\",\"States\":{\"Distributed Map State\":{\"Type\":\"Map\",\"ResultPath\":null,\"End\":true,\"ItemProcessor\":{\"ProcessorConfig\":{\"Mode\":\"DISTRIBUTED\",\"ExecutionType\":\"EXPRESS\"},\"StartAt\":\"Pass State\",\"States\":{\"Pass State\":{\"Type\":\"Pass\",\"End\":true}}},\"MaxConcurrency\":1000,\"ItemReader\":{\"Resource\":\"arn:",
       {
        "Ref": "AWS::Partition"
       },
       ":states:::s3:getObject\",\"ReaderConfig\":{\"InputType\":\"CSV\",\"CSVHeaderLocation\":\"FIRST_ROW\"},\"Parameters\":{\"Bucket\":\"",
       {
        "Ref": "Bucket2001301D4930"
       },
       "\",\"Key.$\":\"$.s3Key\"}}}}}"
      ]
     ]
    },

You could try using Escape hatches for workaround. Marking the issue as appropriate.

[mazyu36]

I encountered this problem before , and I’m working on it.

I think the executionType property in the itemProcessor config is not needed to set when use the DistributedMap Construct.

The mapExecutionType in the DistributedMap Construct is used to set the Processing Mode of the ItemProcessor, and its purpose overlaps with the executionType property in the ItemProcessor Config.

In the current implementation, the mapExecutionType in the DistributedMap Construct always overwrites the Processing Mode of the ItemProcessor.
So the executionType property in the itemProcessor config is meaningless.

The mapExecutionType in the DistributedMap Construct may have been an unnecessary property, but removing it would cause a breaking change, so it should be avoided.