fix(stepfunctions): the catch field in CustomState is not rendered (#29654)

### Issue # (if applicable)

N/A

### Reason for this change

Customers that specify `Catch` fields in their CustomState's `stateJson` do not have Catchers defined in the rendered state definition. The reason for this is that the `Catch` fields from the `stateJson` is overridden by Catchers added through `addCatch()`.

### Description of changes

This change updates the way the state's `Catch` field is rendered to merge Catchers provided inline with those provided through `addCatch()`. Catchers from `addCatch()` will be rendered first, followed by those provided inline. This is consistent with the merge behaviour for Retriers. 

### Description of how you validated changes

Unit test coverage for Catchers provided just inline, just through `addCatch()`, and a combination of both. 

### Checklist
- [X] My code adheres to the [CONTRIBUTING GUIDE](https://github.com/aws/aws-cdk/blob/main/CONTRIBUTING.md) and [DESIGN GUIDELINES](https://github.com/aws/aws-cdk/blob/main/docs/DESIGN_GUIDELINES.md)

----

*By submitting this pull request, I confirm that my contribution is made under the terms of the Apache-2.0 license*

Author: jormello

packages/@aws-cdk-testing/framework-integ/test/aws-stepfunctions/test/integ.custom-state.js.snapshot/aws-stepfunctions-custom-state-integ.template.json

@@ -20,7 +20,7 @@
   "StateMachine2E01A3A5": {
    "Type": "AWS::StepFunctions::StateMachine",
    "Properties": {
-    "DefinitionString": "{\"StartAt\":\"my custom task\",\"States\":{\"my custom task\":{\"Next\":\"my custom task with inline Retriers\",\"Type\":\"Task\",\"Resource\":\"arn:aws:states:::dynamodb:putItem\",\"Parameters\":{\"TableName\":\"my-cool-table\",\"Item\":{\"id\":{\"S\":\"my-entry\"}}},\"ResultPath\":null,\"Retry\":[{\"ErrorEquals\":[\"States.Timeout\"],\"IntervalSeconds\":10,\"MaxAttempts\":5},{\"ErrorEquals\":[\"States.Permissions\"],\"IntervalSeconds\":20,\"MaxAttempts\":2}],\"Catch\":[{\"ErrorEquals\":[\"States.ALL\"],\"Next\":\"failed\"}]},\"my custom task with inline Retriers\":{\"Next\":\"final step\",\"Type\":\"Task\",\"Resource\":\"arn:aws:states:::dynamodb:putItem\",\"Parameters\":{\"TableName\":\"my-cool-table\",\"Item\":{\"id\":{\"S\":\"my-entry\"}}},\"ResultPath\":null,\"Retry\":[{\"ErrorEquals\":[\"States.Permissions\"],\"IntervalSeconds\":20,\"MaxAttempts\":2}]},\"final step\":{\"Type\":\"Pass\",\"End\":true},\"failed\":{\"Type\":\"Fail\",\"Error\":\"DidNotWork\",\"Cause\":\"We got stuck\"}},\"TimeoutSeconds\":30}",
+    "DefinitionString": "{\"StartAt\":\"my custom task\",\"States\":{\"my custom task\":{\"Next\":\"my custom task with inline Retriers\",\"Type\":\"Task\",\"Resource\":\"arn:aws:states:::dynamodb:putItem\",\"Parameters\":{\"TableName\":\"my-cool-table\",\"Item\":{\"id\":{\"S\":\"my-entry\"}}},\"ResultPath\":null,\"Retry\":[{\"ErrorEquals\":[\"States.Timeout\"],\"IntervalSeconds\":10,\"MaxAttempts\":5}],\"Catch\":[{\"ErrorEquals\":[\"States.ALL\"],\"Next\":\"failed\"}]},\"my custom task with inline Retriers\":{\"Next\":\"my custom task with inline Catchers\",\"Type\":\"Task\",\"Resource\":\"arn:aws:states:::dynamodb:putItem\",\"Parameters\":{\"TableName\":\"my-cool-table\",\"Item\":{\"id\":{\"S\":\"my-entry\"}}},\"ResultPath\":null,\"Retry\":[{\"ErrorEquals\":[\"States.Permissions\"],\"IntervalSeconds\":20,\"MaxAttempts\":2}]},\"my custom task with inline Catchers\":{\"Next\":\"final step\",\"Type\":\"Task\",\"Resource\":\"arn:aws:states:::dynamodb:putItem\",\"Parameters\":{\"TableName\":\"my-cool-table\",\"Item\":{\"id\":{\"S\":\"my-entry\"}}},\"ResultPath\":null,\"Catch\":[{\"ErrorEquals\":[\"States.Permissions\"],\"Next\":\"failed\"}]},\"final step\":{\"Type\":\"Pass\",\"End\":true},\"failed\":{\"Type\":\"Fail\",\"Error\":\"DidNotWork\",\"Cause\":\"We got stuck\"}},\"TimeoutSeconds\":30}",
     "RoleArn": {
      "Fn::GetAtt": [
       "StateMachineRoleB840431D",

packages/@aws-cdk-testing/framework-integ/test/aws-stepfunctions/test/integ.custom-state.ts

@@ -23,11 +23,6 @@ const stateJson = {
     },
   },
   ResultPath: null,
-  Retry: [{
-    ErrorEquals: [sfn.Errors.PERMISSIONS],
-    IntervalSeconds: 20,
-    MaxAttempts: 2,
-  }],
 };
 
 const failure = new sfn.Fail(stack, 'failed', {
@@ -39,18 +34,35 @@ const custom = new sfn.CustomState(stack, 'my custom task', {
   stateJson,
 });
 
-const customWithInlineRetry = new sfn.CustomState(stack, 'my custom task with inline Retriers', {
-  stateJson,
-});
-
 custom.addCatch(failure);
 custom.addRetry({
   errors: [sfn.Errors.TIMEOUT],
   interval: cdk.Duration.seconds(10),
   maxAttempts: 5,
 });
 
-const chain = sfn.Chain.start(custom).next(customWithInlineRetry).next(finalStatus);
+const customWithInlineRetry = new sfn.CustomState(stack, 'my custom task with inline Retriers', {
+  stateJson: {
+    ...stateJson,
+    Retry: [{
+      ErrorEquals: [sfn.Errors.PERMISSIONS],
+      IntervalSeconds: 20,
+      MaxAttempts: 2,
+    }],
+  },
+});
+
+const customWithInlineCatch = new sfn.CustomState(stack, 'my custom task with inline Catchers', {
+  stateJson: {
+    ...stateJson,
+    Catch: [{
+      ErrorEquals: [sfn.Errors.PERMISSIONS],
+      Next: 'failed',
+    }],
+  },
+});
+
+const chain = sfn.Chain.start(custom).next(customWithInlineRetry).next(customWithInlineCatch).next(finalStatus);
 
 const sm = new sfn.StateMachine(stack, 'StateMachine', {
   definition: chain,

packages/aws-cdk-lib/aws-stepfunctions/lib/states/custom-state.ts

@@ -1,6 +1,7 @@
 import { Construct } from 'constructs';
 import { State } from './state';
 import { Chain } from '..';
+import { Annotations } from '../../../core/';
 import { CatchProps, IChainable, INextable, RetryProps } from '../types';
 
 /**
@@ -74,11 +75,64 @@ export class CustomState extends State implements IChainable, INextable {
       ...this.renderRetryCatch(),
     };
 
-    // merge the Retry field defined in the stateJson into the state
+    if (this.hasMultipleRetrySources(state)) {
+      this.addMultipleRetrySourcesWarning();
+    }
+
+    if (this.hasMultipleCatchSources(state)) {
+      this.addMultipleCatchSourcesWarning();
+    }
+
+    // Retriers and Catchers can be specified directly in the stateJson or indirectly to the construct with addRetry() and addCatch().
+    // renderRetryCatch() only renders the indirectly supplied Retriers and Catchers, so we need to manually merge in those directly in the stateJson
     if (Array.isArray(this.stateJson.Retry)) {
       state.Retry = Array.isArray(state.Retry) ? [...state.Retry, ...this.stateJson.Retry] : [...this.stateJson.Retry];
     }
 
+    if (Array.isArray(this.stateJson.Catch)) {
+      state.Catch = Array.isArray(state.Catch) ? [...state.Catch, ...this.stateJson.Catch] : [...this.stateJson.Catch];
+    }
+
     return state;
   }
+
+  private hasMultipleRetrySources(state: any): boolean {
+    if (!Array.isArray(state.Retry)) {
+      return false;
+    }
+
+    if (!Array.isArray(this.stateJson.Retry)) {
+      return false;
+    }
+
+    return state.Retry.length > 0 && this.stateJson.Retry.length > 0;
+  }
+
+  private hasMultipleCatchSources(state: any): boolean {
+    if (!Array.isArray(state.Catch)) {
+      return false;
+    }
+
+    if (!Array.isArray(this.stateJson.Catch)) {
+      return false;
+    }
+
+    return state.Catch.length > 0 && this.stateJson.Catch.length > 0;
+  }
+
+  private addMultipleRetrySourcesWarning(): void {
+    Annotations.of(this).addWarningV2('@aws-cdk/aws-stepfunctions:multipleRetrySources', [
+      'CustomState constructs can configure state retries using the stateJson property or by using the addRetry() function.',
+      'When retries are configured using both of these, the state definition\'s Retry field is generated ',
+      'by first rendering retries from addRetry(), then rendering retries from the stateJson.',
+    ].join('\n'));
+  }
+
+  private addMultipleCatchSourcesWarning(): void {
+    Annotations.of(this).addWarningV2('@aws-cdk/aws-stepfunctions:multipleCatchSources', [
+      'CustomState constructs can configure state catchers using the stateJson property or by using the addCatch() function.',
+      'When catchers are configured using both of these, the state definition\'s Catch field is generated ',
+      'by first rendering catchers from addCatch(), then rendering catchers from the stateJson.',
+    ].join('\n'));
+  }
 }

packages/aws-cdk-lib/aws-stepfunctions/test/custom-state.test.ts

@@ -1,6 +1,8 @@
 import { render } from './private/render-util';
+import { Annotations, Match } from '../../assertions';
 import * as cdk from '../../core';
 import * as sfn from '../lib';
+import { Errors } from '../lib/types';
 
 describe('Custom State', () => {
   let stack: cdk.Stack;
@@ -309,4 +311,215 @@ describe('Custom State', () => {
       },
     );
   });
+
+  test('expect retry to merge when specifying strategy inline and through construct', () => {
+    // GIVEN
+    const custom = new sfn.CustomState(stack, 'Custom', {
+      stateJson: {
+        ...stateJson,
+        Retry: [{
+          ErrorEquals: ['States.TaskFailed'],
+        }],
+      },
+    }).addRetry({ errors: [Errors.TIMEOUT] });
+    const chain = sfn.Chain.start(custom);
+
+    // THEN
+    expect(render(stack, chain)).toStrictEqual(
+      {
+        StartAt: 'Custom',
+        States: {
+          Custom: {
+            Type: 'Task',
+            Resource: 'arn:aws:states:::dynamodb:putItem',
+            Parameters: {
+              TableName: 'MyTable',
+              Item: {
+                id: {
+                  S: 'MyEntry',
+                },
+              },
+            },
+            ResultPath: null,
+            Retry: [
+              {
+                ErrorEquals: ['States.Timeout'],
+              },
+              {
+                ErrorEquals: ['States.TaskFailed'],
+              },
+            ],
+            End: true,
+          },
+        },
+      },
+    );
+  });
+
+  test('expect catch to not fail when specifying strategy inline', () => {
+    // GIVEN
+    const custom = new sfn.CustomState(stack, 'Custom', {
+      stateJson: {
+        ...stateJson,
+        Catch: [{
+          ErrorEquals: ['States.TaskFailed'],
+          Next: 'Failed',
+        }],
+      },
+    });
+    const chain = sfn.Chain.start(custom);
+
+    // THEN
+    expect(render(stack, chain)).toStrictEqual(
+      {
+        StartAt: 'Custom',
+        States: {
+          Custom: {
+            Type: 'Task',
+            Resource: 'arn:aws:states:::dynamodb:putItem',
+            Parameters: {
+              TableName: 'MyTable',
+              Item: {
+                id: {
+                  S: 'MyEntry',
+                },
+              },
+            },
+            ResultPath: null,
+            Catch: [{
+              ErrorEquals: ['States.TaskFailed'],
+              Next: 'Failed',
+            }],
+            End: true,
+          },
+        },
+      },
+    );
+  });
+
+  test('expect catch to merge when specifying strategy inline and through construct', () => {
+    // GIVEN
+    const failure = new sfn.Fail(stack, 'Failed', {
+      error: 'DidNotWork',
+      cause: 'We got stuck',
+    });
+
+    const custom = new sfn.CustomState(stack, 'Custom', {
+      stateJson: {
+        ...stateJson,
+        Catch: [{
+          ErrorEquals: ['States.TaskFailed'],
+          Next: 'Failed',
+        }],
+      },
+    }).addCatch(failure, { errors: [Errors.TIMEOUT] });
+    const chain = sfn.Chain.start(custom);
+
+    // THEN
+    expect(render(stack, chain)).toStrictEqual(
+      {
+        StartAt: 'Custom',
+        States: {
+          Custom: {
+            Type: 'Task',
+            Resource: 'arn:aws:states:::dynamodb:putItem',
+            Parameters: {
+              TableName: 'MyTable',
+              Item: {
+                id: {
+                  S: 'MyEntry',
+                },
+              },
+            },
+            ResultPath: null,
+            Catch: [
+              {
+                ErrorEquals: ['States.Timeout'],
+                Next: 'Failed',
+              }, {
+                ErrorEquals: ['States.TaskFailed'],
+                Next: 'Failed',
+              },
+            ],
+            End: true,
+          },
+          Failed: {
+            Type: 'Fail',
+            Error: 'DidNotWork',
+            Cause: 'We got stuck',
+          },
+        },
+      },
+    );
+  });
+
+  test('expect warning message to be emitted when retries specified both in stateJson and through addRetry()', () => {
+    const customState = new sfn.CustomState(stack, 'my custom task', {
+      stateJson: {
+        Type: 'Task',
+        Resource: 'arn:aws:states:::dynamodb:putItem',
+        Parameters: {
+          TableName: 'my-cool-table',
+          Item: {
+            id: {
+              S: 'my-entry',
+            },
+          },
+        },
+        Retry: [{
+          ErrorEquals: ['States.TaskFailed'],
+        }],
+      },
+    });
+
+    customState.addRetry({
+      errors: [sfn.Errors.TIMEOUT],
+      interval: cdk.Duration.seconds(10),
+      maxAttempts: 5,
+    });
+
+    new sfn.StateMachine(stack, 'StateMachine', {
+      definition: sfn.Chain.start(customState),
+      timeout: cdk.Duration.seconds(30),
+    });
+
+    Annotations.fromStack(stack).hasWarning('/Default/my custom task', Match.stringLikeRegexp('CustomState constructs can configure state retries'));
+  });
+
+  test('expect warning message to be emitted when catchers specified both in stateJson and through addCatch()', () => {
+    const customState = new sfn.CustomState(stack, 'my custom task', {
+      stateJson: {
+        Type: 'Task',
+        Resource: 'arn:aws:states:::dynamodb:putItem',
+        Parameters: {
+          TableName: 'my-cool-table',
+          Item: {
+            id: {
+              S: 'my-entry',
+            },
+          },
+        },
+        Catch: [
+          {
+            ErrorEquals: ['States.Timeout'],
+            Next: 'Failed',
+          },
+        ],
+      },
+    });
+
+    const failure = new sfn.Fail(stack, 'Failed', {
+      error: 'DidNotWork',
+      cause: 'We got stuck',
+    });
+
+    customState.addCatch(failure, { errors: [Errors.TIMEOUT] });
+
+    new sfn.StateMachine(stack, 'StateMachine', {
+      definition: sfn.Chain.start(customState),
+      timeout: cdk.Duration.seconds(30),
+    });
+
+    Annotations.fromStack(stack).hasWarning('/Default/my custom task', Match.stringLikeRegexp('CustomState constructs can configure state catchers'));
+  });
 });