fix(stepfunctions-tasks): specify tags in BatchSubmitJob properties

packages/@aws-cdk-testing/framework-integ/test/aws-stepfunctions-tasks/test/batch/integ.submit-job.js.snapshot/aws-stepfunctions-integ.template.json

@@ -758,7 +758,7 @@
          "JobQueueArn"
         ]
        },
-       "\",\"Parameters\":{\"foo.$\":\"$.bar\"},\"ContainerOverrides\":{\"Environment\":[{\"Name\":\"key\",\"Value\":\"value\"}],\"ResourceRequirements\":[{\"Type\":\"MEMORY\",\"Value\":\"256\"},{\"Type\":\"VCPU\",\"Value\":\"1\"}]},\"RetryStrategy\":{\"Attempts\":3},\"Timeout\":{\"AttemptDurationSeconds\":60}}}}}"
+       "\",\"Parameters\":{\"foo.$\":\"$.bar\"},\"ContainerOverrides\":{\"Environment\":[{\"Name\":\"key\",\"Value\":\"value\"}],\"ResourceRequirements\":[{\"Type\":\"MEMORY\",\"Value\":\"256\"},{\"Type\":\"VCPU\",\"Value\":\"1\"}]},\"RetryStrategy\":{\"Attempts\":3},\"Tags\":{\"key\":\"value\"},\"Timeout\":{\"AttemptDurationSeconds\":60}}}}}"
       ]
      ]
     }

packages/@aws-cdk-testing/framework-integ/test/aws-stepfunctions-tasks/test/batch/integ.submit-job.ts

@@ -56,6 +56,9 @@ class RunBatchStack extends cdk.Stack {
       }),
       attempts: 3,
       taskTimeout: sfn.Timeout.duration(cdk.Duration.seconds(60)),
+      tags: {
+        key: 'value',
+      },
     });
 
     const definition = new sfn.Pass(this, 'Start', {

packages/@aws-cdk/aws-batch-alpha/test/aws-stepfunctions-tasks/submit-job.test.ts

@@ -361,3 +361,89 @@ test('Task throws if attempt duration is less than 60 sec', () => {
     /attempt duration must be greater than 60 seconds./,
   );
 });
+
+test('supports passing tags', () => {
+  // WHEN
+  const task = new BatchSubmitJob(stack, 'Task', {
+    jobDefinitionArn: batchJobDefinition.jobDefinitionArn,
+    jobQueueArn: batchJobQueue.jobQueueArn,
+    jobName: sfn.JsonPath.stringAt('$.jobName'),
+    tags: {
+      test: 'this is a tag',
+    },
+  });
+
+  // THEN
+  expect(stack.resolve(task.toStateJson())).toEqual({
+    Type: 'Task',
+    Resource: {
+      'Fn::Join': [
+        '',
+        [
+          'arn:',
+          {
+            Ref: 'AWS::Partition',
+          },
+          ':states:::batch:submitJob.sync',
+        ],
+      ],
+    },
+    End: true,
+    Parameters: {
+      'JobDefinition': { Ref: 'JobDefinition24FFE3ED' },
+      'JobName.$': '$.jobName',
+      'JobQueue': {
+        'Fn::GetAtt': [
+          'JobQueueEE3AD499',
+          'JobQueueArn',
+        ],
+      },
+      'Tags': {
+        test: 'this is a tag',
+      },
+    },
+  });
+});
+
+test('throws if tags has invalid value', () => {
+  expect(() => {
+    const tags: { [key: string]: string } = {};
+    for (let i = 0; i < 100; i++) {
+      tags[i] = 'tag';
+    }
+    new BatchSubmitJob(stack, 'Task1', {
+      jobDefinitionArn: batchJobDefinition.jobDefinitionArn,
+      jobName: 'JobName',
+      jobQueueArn: batchJobQueue.jobQueueArn,
+      tags,
+    });
+  }).toThrow(
+    /Maximum tag number of entries is 50./,
+  );
+
+  expect(() => {
+    const keyTooLong = 'k'.repeat(150);
+    const tags: { [key: string]: string } = {};
+    tags[keyTooLong] = 'tag';
+    new BatchSubmitJob(stack, 'Task2', {
+      jobDefinitionArn: batchJobDefinition.jobDefinitionArn,
+      jobName: 'JobName',
+      jobQueueArn: batchJobQueue.jobQueueArn,
+      tags,
+    });
+  }).toThrow(
+    /Tag key size must be between 1 and 128/,
+  );
+
+  expect(() => {
+    const tags = { key: 'k'.repeat(300) };
+    new BatchSubmitJob(stack, 'Task3', {
+      jobDefinitionArn: batchJobDefinition.jobDefinitionArn,
+      jobName: 'JobName',
+      jobQueueArn: batchJobQueue.jobQueueArn,
+      tags,
+    });
+  }).toThrow(
+    /Tag value maximum size is 256/,
+  );
+});

packages/aws-cdk-lib/aws-stepfunctions-tasks/lib/batch/submit-job.ts

@@ -147,6 +147,13 @@ export interface BatchSubmitJobProps extends sfn.TaskStateBaseProps {
    * @default 1
    */
   readonly attempts?: number;
+
+  /**
+   * The tags applied to the job request.
+   *
+   * @default {} - no tags
+   */
+  readonly tags?: { [key: string]: string };
 }
 
 /**
@@ -212,6 +219,8 @@ export class BatchSubmitJob extends sfn.TaskStateBase {
       });
     }
 
+    this.validateTags(props.tags);
+
     this.taskPolicies = this.configurePolicyStatements();
   }
 
@@ -255,7 +264,7 @@ export class BatchSubmitJob extends sfn.TaskStateBase {
           this.props.attempts !== undefined
             ? { Attempts: this.props.attempts }
             : undefined,
-
+        Tags: this.props.tags,
         Timeout: timeout
           ? { AttemptDurationSeconds: timeout }
           : undefined,
@@ -337,4 +346,20 @@ export class BatchSubmitJob extends sfn.TaskStateBase {
       ResourceRequirements: resources.length ? resources : undefined,
     };
   }
+
+  private validateTags(tags?: { [key: string]: string }) {
+    if (tags === undefined) return;
+    const tagEntries = Object.entries(tags);
+    if (tagEntries.length > 50) {
+      throw new Error(`Maximum tag number of entries is 50. Received ${tagEntries.length}.`);
+    }
+    for (const [key, value] of tagEntries) {
+      if (key.length < 1 || key.length > 128) {
+        throw new Error(`Tag key size must be between 1 and 128, but got ${key.length}.`);
+      }
+      if (value.length > 256) {
+        throw new Error(`Tag value maximum size is 256, but got ${value.length}.`);
+      }
+    }
+  }
 }