feat(stepfunctions-tasks): add 'Emr on Eks' tasks (#17103)

This CDK feature adds support for Emr on Eks by implementing API service integrations for the following three APIs.

This PR adds three tasks which support Emr on Eks:
1) [Create Virtual Cluster](https://docs.aws.amazon.com/emr-on-eks/latest/APIReference/API_CreateVirtualCluster.html)
2) [ Start a job run](https://docs.aws.amazon.com/emr-on-eks/latest/APIReference/API_StartJobRun.html)
3) [Delete virtual cluster ](https://docs.aws.amazon.com/emr-on-eks/latest/APIReference/API_DeleteVirtualCluster.html)


Continuation of #15262 by @matthewsvu and @BenChaimberg:

Closes #15234.

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

Author: sanava2010

packages/@aws-cdk/aws-stepfunctions-tasks/README.md

@@ -60,6 +60,10 @@ This module is part of the [AWS Cloud Development Kit](https://github.com/aws/aw
     - [Cancel Step](#cancel-step)
     - [Modify Instance Fleet](#modify-instance-fleet)
     - [Modify Instance Group](#modify-instance-group)
+  - [EMR on EKS](#emr-on-eks)
+    - [Create Virtual Cluster](#create-virtual-cluster)
+    - [Delete Virtual Cluster](#delete-virtual-cluster)
+    - [Start Job Run](#start-job-run)
   - [EKS](#eks)
     - [Call](#call)
   - [EventBridge](#eventbridge)
@@ -783,6 +787,169 @@ new tasks.EmrModifyInstanceGroupByName(this, 'Task', {
 });
 ```
 
+## EMR on EKS
+
+Step Functions supports Amazon EMR on EKS through the service integration pattern.
+The service integration APIs correspond to Amazon EMR on EKS APIs, but differ in the parameters that are used.
+
+[Read more](https://docs.aws.amazon.com/step-functions/latest/dg/connect-emr-eks.html) about the differences when using these service integrations.
+
+[Setting up](https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/setting-up.html) the EKS cluster is required.
+
+### Create Virtual Cluster
+
+The [CreateVirtualCluster](https://docs.aws.amazon.com/emr-on-eks/latest/APIReference/API_CreateVirtualCluster.html) API creates a single virtual cluster that's mapped to a single Kubernetes namespace. 
+
+The EKS cluster containing the Kubernetes namespace where the virtual cluster will be mapped can be passed in from the task input.
+
+```ts
+new tasks.EmrContainersCreateVirtualCluster(this, 'Create a Virtual Cluster', {
+  eksCluster: tasks.EksClusterInput.fromTaskInput(sfn.TaskInput.fromText('clusterId')),
+});
+```
+
+The EKS cluster can also be passed in directly.
+
+```ts
+import * as eks from '@aws-cdk/aws-eks';
+
+declare const eksCluster: eks.Cluster;
+
+new tasks.EmrContainersCreateVirtualCluster(this, 'Create a Virtual Cluster', {
+  eksCluster: tasks.EksClusterInput.fromCluster(eksCluster),
+});
+```
+
+By default, the Kubernetes namespace that a virtual cluster maps to is "default", but a specific namespace within an EKS cluster can be selected.
+
+```ts
+new tasks.EmrContainersCreateVirtualCluster(this, 'Create a Virtual Cluster', {
+  eksCluster: tasks.EksClusterInput.fromTaskInput(sfn.TaskInput.fromText('clusterId')),
+  eksNamespace: 'specified-namespace',
+});
+```
+
+### Delete Virtual Cluster
+
+The [DeleteVirtualCluster](https://docs.aws.amazon.com/emr-on-eks/latest/APIReference/API_DeleteVirtualCluster.html) API deletes a virtual cluster.
+
+```ts
+new tasks.EmrContainersDeleteVirtualCluster(this, 'Delete a Virtual Cluster', {
+  virtualClusterId: sfn.TaskInput.fromJsonPathAt('$.virtualCluster'),
+});
+```
+
+### Start Job Run
+
+The [StartJobRun](https://docs.aws.amazon.com/emr-on-eks/latest/APIReference/API_StartJobRun.html) API starts a job run. A job is a unit of work that you submit to Amazon EMR on EKS for execution. The work performed by the job can be defined by a Spark jar, PySpark script, or SparkSQL query. A job run is an execution of the job on the virtual cluster.
+
+Required setup:
+
+ - If not done already, follow the [steps](https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/setting-up.html) to setup EMR on EKS and [create an EKS Cluster](https://docs.aws.amazon.com/cdk/api/latest/docs/aws-eks-readme.html#quick-start).
+ - Enable [Cluster access](https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/setting-up-cluster-access.html)
+ - Enable [IAM Role access](https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/setting-up-enable-IAM.html)
+
+The following actions must be performed if the virtual cluster ID is supplied from the task input. Otherwise, if it is supplied statically in the state machine definition, these actions will be done automatically.
+
+ - Create an [IAM role](https://docs.aws.amazon.com/cdk/api/latest/docs/@aws-cdk_aws-iam.Role.html)
+ - Update the [Role Trust Policy](https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/setting-up-trust-policy.html) of the Job Execution Role.
+
+The job can be configured with spark submit parameters:
+
+```ts
+new tasks.EmrContainersStartJobRun(this, 'EMR Containers Start Job Run', {
+  virtualCluster: tasks.VirtualClusterInput.fromVirtualClusterId('de92jdei2910fwedz'),
+  releaseLabel: tasks.ReleaseLabel.EMR_6_2_0,
+  jobDriver: {
+    sparkSubmitJobDriver: {
+      entryPoint: sfn.TaskInput.fromText('local:///usr/lib/spark/examples/src/main/python/pi.py'),
+      sparkSubmitParameters: '--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1',
+    },
+  },
+});
+```
+
+Configuring the job can also be done via application configuration:
+
+```ts
+new tasks.EmrContainersStartJobRun(this, 'EMR Containers Start Job Run', {
+  virtualCluster: tasks.VirtualClusterInput.fromVirtualClusterId('de92jdei2910fwedz'),
+  releaseLabel: tasks.ReleaseLabel.EMR_6_2_0,
+  jobName: 'EMR-Containers-Job',
+  jobDriver: {
+    sparkSubmitJobDriver: {
+      entryPoint: sfn.TaskInput.fromText('local:///usr/lib/spark/examples/src/main/python/pi.py'),
+    },
+  },
+  applicationConfig: [{
+    classification: tasks.Classification.SPARK_DEFAULTS,
+    properties: {
+      'spark.executor.instances': '1',
+      'spark.executor.memory': '512M',
+    },
+  }],
+});
+```
+
+Job monitoring can be enabled if `monitoring.logging` is set true. This automatically generates an S3 bucket and CloudWatch logs.
+
+```ts
+new tasks.EmrContainersStartJobRun(this, 'EMR Containers Start Job Run', {
+  virtualCluster: tasks.VirtualClusterInput.fromVirtualClusterId('de92jdei2910fwedz'),
+  releaseLabel: tasks.ReleaseLabel.EMR_6_2_0,
+  jobDriver: {
+    sparkSubmitJobDriver: {
+      entryPoint: sfn.TaskInput.fromText('local:///usr/lib/spark/examples/src/main/python/pi.py'),
+      sparkSubmitParameters: '--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1',
+    },
+  },
+  monitoring: {
+    logging: true,
+  },
+});
+```
+
+Otherwise, providing monitoring for jobs with existing log groups and log buckets is also available.
+
+```ts
+import * as logs from '@aws-cdk/aws-logs';
+
+const logGroup = new logs.LogGroup(this, 'Log Group');
+const logBucket = new s3.Bucket(this, 'S3 Bucket')
+
+new tasks.EmrContainersStartJobRun(this, 'EMR Containers Start Job Run', {
+  virtualCluster: tasks.VirtualClusterInput.fromVirtualClusterId('de92jdei2910fwedz'),
+  releaseLabel: tasks.ReleaseLabel.EMR_6_2_0,
+  jobDriver: {
+    sparkSubmitJobDriver: {
+      entryPoint: sfn.TaskInput.fromText('local:///usr/lib/spark/examples/src/main/python/pi.py'),
+      sparkSubmitParameters: '--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1',
+    },
+  },
+  monitoring: {
+    logGroup: logGroup,
+    logBucket: logBucket,
+  },
+});
+```
+
+Users can provide their own existing Job Execution Role.
+
+```ts
+new tasks.EmrContainersStartJobRun(this, 'EMR Containers Start Job Run', {
+  virtualCluster:tasks.VirtualClusterInput.fromTaskInput(sfn.TaskInput.fromJsonPathAt('$.VirtualClusterId')),
+  releaseLabel: tasks.ReleaseLabel.EMR_6_2_0,
+  jobName: 'EMR-Containers-Job',
+  executionRole: iam.Role.fromRoleArn(this, 'Job-Execution-Role', 'arn:aws:iam::xxxxxxxxxxxx:role/JobExecutionRole'),
+  jobDriver: {
+    sparkSubmitJobDriver: {
+      entryPoint: sfn.TaskInput.fromText('local:///usr/lib/spark/examples/src/main/python/pi.py'),
+      sparkSubmitParameters: '--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1',
+    },
+  },
+});
+```
+
 ## EKS
 
 Step Functions supports Amazon EKS through the service integration pattern.

packages/@aws-cdk/aws-stepfunctions-tasks/lib/emrcontainers/create-virtual-cluster.ts

@@ -0,0 +1,146 @@
+import * as eks from '@aws-cdk/aws-eks';
+import * as iam from '@aws-cdk/aws-iam';
+import * as sfn from '@aws-cdk/aws-stepfunctions';
+import { Stack } from '@aws-cdk/core';
+import { Construct } from 'constructs';
+import { integrationResourceArn, validatePatternSupported } from '../private/task-utils';
+
+/**
+ * Class for supported types of EMR Containers' Container Providers
+ */
+enum ContainerProviderTypes {
+
+  /**
+   * Supported container provider type for a EKS Cluster
+   */
+  EKS = 'EKS'
+}
+
+/**
+ * Class that supports methods which return the EKS cluster name depending on input type.
+ */
+export class EksClusterInput {
+
+  /**
+   * Specify an existing EKS Cluster as the name for this Cluster
+   */
+  static fromCluster(cluster: eks.ICluster): EksClusterInput {
+    return new EksClusterInput(cluster.clusterName);
+  }
+
+  /**
+   * Specify a Task Input as the name for this Cluster
+   */
+  static fromTaskInput(taskInput: sfn.TaskInput): EksClusterInput {
+    return new EksClusterInput(taskInput.value);
+  }
+
+  /**
+   * Initializes the clusterName
+   *
+   * @param clusterName The name of the EKS Cluster
+   */
+  private constructor(readonly clusterName: string) { }
+}
+
+/**
+ * Properties to define a EMR Containers CreateVirtualCluster Task on an EKS cluster
+ */
+export interface EmrContainersCreateVirtualClusterProps extends sfn.TaskStateBaseProps {
+
+  /**
+   * EKS Cluster or task input that contains the name of the cluster
+   */
+  readonly eksCluster: EksClusterInput;
+
+  /**
+   * The namespace of an EKS cluster
+   *
+   * @default - 'default'
+   */
+  readonly eksNamespace?: string;
+
+  /**
+   * Name of the virtual cluster that will be created.
+   *
+   * @default - the name of the state machine execution that runs this task and state name
+   */
+  readonly virtualClusterName?: string;
+
+  /**
+   * The tags assigned to the virtual cluster
+   *
+   * @default {}
+   */
+  readonly tags?: { [key: string]: string };
+}
+
+/**
+ * Task that creates an EMR Containers virtual cluster from an EKS cluster
+ *
+ * @see https://docs.aws.amazon.com/step-functions/latest/dg/connect-emr-eks.html
+ */
+export class EmrContainersCreateVirtualCluster extends sfn.TaskStateBase {
+
+  private static readonly SUPPORTED_INTEGRATION_PATTERNS: sfn.IntegrationPattern[] = [
+    sfn.IntegrationPattern.REQUEST_RESPONSE,
+  ];
+
+  protected readonly taskMetrics?: sfn.TaskMetricsConfig;
+  protected readonly taskPolicies?: iam.PolicyStatement[];
+
+  private readonly integrationPattern: sfn.IntegrationPattern;
+
+  constructor(scope: Construct, id: string, private readonly props: EmrContainersCreateVirtualClusterProps) {
+    super(scope, id, props);
+    this.integrationPattern = props.integrationPattern ?? sfn.IntegrationPattern.REQUEST_RESPONSE;
+    validatePatternSupported(this.integrationPattern, EmrContainersCreateVirtualCluster.SUPPORTED_INTEGRATION_PATTERNS);
+
+    this.taskPolicies = this.createPolicyStatements();
+  }
+
+  /**
+   * @internal
+   */
+  protected _renderTask(): any {
+    return {
+      Resource: integrationResourceArn('emr-containers', 'createVirtualCluster', this.integrationPattern),
+      Parameters: sfn.FieldUtils.renderObject({
+        Name: this.props.virtualClusterName ?? sfn.JsonPath.stringAt('States.Format(\'{}/{}\', $$.Execution.Name, $$.State.Name)'),
+        ContainerProvider: {
+          Id: this.props.eksCluster.clusterName,
+          Info: {
+            EksInfo: {
+              Namespace: this.props.eksNamespace ?? 'default',
+            },
+          },
+          Type: ContainerProviderTypes.EKS,
+        },
+        Tags: this.props.tags,
+      }),
+    };
+  };
+
+  private createPolicyStatements(): iam.PolicyStatement[] {
+    return [
+      new iam.PolicyStatement({
+        resources: ['*'], // We need * permissions for creating a virtual cluster https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/setting-up-iam.html
+        actions: ['emr-containers:CreateVirtualCluster'],
+      }),
+      new iam.PolicyStatement({
+        resources: [
+          Stack.of(this).formatArn({
+            service: 'iam',
+            region: '',
+            resource: 'role/aws-service-role/emr-containers.amazonaws.com',
+            resourceName: 'AWSServiceRoleForAmazonEMRContainers',
+          }),
+        ],
+        actions: ['iam:CreateServiceLinkedRole'],
+        conditions: {
+          StringLike: { 'iam:AWSServiceName': 'emr-containers.amazonaws.com' },
+        },
+      }),
+    ];
+  }
+}

packages/@aws-cdk/aws-stepfunctions-tasks/lib/emrcontainers/delete-virtual-cluster.ts

@@ -0,0 +1,74 @@
+import * as iam from '@aws-cdk/aws-iam';
+import * as sfn from '@aws-cdk/aws-stepfunctions';
+import * as cdk from '@aws-cdk/core';
+import { Construct } from 'constructs';
+import { integrationResourceArn, validatePatternSupported } from '../private/task-utils';
+
+/**
+ * Properties to define a EMR Containers DeleteVirtualCluster Task
+ */
+export interface EmrContainersDeleteVirtualClusterProps extends sfn.TaskStateBaseProps {
+
+  /**
+   * The ID of the virtual cluster that will be deleted.
+   */
+  readonly virtualClusterId: sfn.TaskInput;
+}
+
+/**
+ * Deletes an EMR Containers virtual cluster as a Task.
+ *
+ * @see https://docs.amazonaws.cn/en_us/step-functions/latest/dg/connect-emr-eks.html
+ */
+export class EmrContainersDeleteVirtualCluster extends sfn.TaskStateBase {
+
+  private static readonly SUPPORTED_INTEGRATION_PATTERNS: sfn.IntegrationPattern[] = [
+    sfn.IntegrationPattern.REQUEST_RESPONSE,
+    sfn.IntegrationPattern.RUN_JOB,
+  ];
+
+  protected readonly taskMetrics?: sfn.TaskMetricsConfig;
+  protected readonly taskPolicies?: iam.PolicyStatement[];
+
+  private readonly integrationPattern: sfn.IntegrationPattern;
+
+  constructor(scope: Construct, id: string, private readonly props: EmrContainersDeleteVirtualClusterProps) {
+    super(scope, id, props);
+    this.integrationPattern = props.integrationPattern ?? sfn.IntegrationPattern.REQUEST_RESPONSE;
+
+    validatePatternSupported(this.integrationPattern, EmrContainersDeleteVirtualCluster.SUPPORTED_INTEGRATION_PATTERNS);
+
+    this.taskPolicies = this.createPolicyStatements();
+  }
+
+  /**
+   * @internal
+   */
+  protected _renderTask(): any {
+    return {
+      Resource: integrationResourceArn('emr-containers', 'deleteVirtualCluster', this.integrationPattern),
+      Parameters: sfn.FieldUtils.renderObject({
+        Id: this.props.virtualClusterId.value,
+      }),
+    };
+  };
+
+  private createPolicyStatements(): iam.PolicyStatement[] {
+    const actions = ['emr-containers:DeleteVirtualCluster'];
+    if (this.integrationPattern === sfn.IntegrationPattern.RUN_JOB) {
+      actions.push('emr-containers:DescribeVirtualCluster');
+    }
+
+    return [new iam.PolicyStatement({
+      resources: [
+        cdk.Stack.of(this).formatArn({
+          arnFormat: cdk.ArnFormat.SLASH_RESOURCE_SLASH_RESOURCE_NAME,
+          service: 'emr-containers',
+          resource: 'virtualclusters',
+          resourceName: sfn.JsonPath.isEncodedJsonPath(this.props.virtualClusterId.value) ? '*' : this.props.virtualClusterId.value,
+        }),
+      ],
+      actions: actions,
+    })];
+  }
+}

packages/@aws-cdk/aws-stepfunctions-tasks/lib/emrcontainers/start-job-run.ts

@@ -0,0 +1,16 @@
+import subprocess as sp
+import os
+import logging
+
+#https://github.com/aws/aws-cdk/tree/master/packages/%40aws-cdk/aws-stepfunctions#custom-state
+def handler(event, context):
+    logger = logging.getLogger()
+    logger.setLevel(logging.INFO)
+    command = f"/opt/awscli/aws emr-containers update-role-trust-policy --cluster-name {event['ResourceProperties']['eksClusterId']} --namespace {event['ResourceProperties']['eksNamespace']} --role-name {event['ResourceProperties']['roleName']}"
+    if event['RequestType'] == 'Create' or event['RequestType'] == 'Update' :
+        try:
+            res = sp.check_output(command, shell=True)
+            logger.info(f"Successfully ran {command}")
+        except Exception as e:
+            logger.info(f"ERROR: {str(e)}")
+            

packages/@aws-cdk/aws-stepfunctions-tasks/lib/emrcontainers/utils/role-policy/index.py

@@ -29,6 +29,9 @@ export * from './emr/emr-add-step';
 export * from './emr/emr-cancel-step';
 export * from './emr/emr-modify-instance-fleet-by-name';
 export * from './emr/emr-modify-instance-group-by-name';
+export * from './emrcontainers/create-virtual-cluster';
+export * from './emrcontainers/delete-virtual-cluster';
+export * from './emrcontainers/start-job-run';
 export * from './glue/run-glue-job-task';
 export * from './glue/start-job-run';
 export * from './batch/run-batch-job';

packages/@aws-cdk/aws-stepfunctions-tasks/lib/index.ts

@@ -107,11 +107,14 @@
     "@aws-cdk/aws-iam": "0.0.0",
     "@aws-cdk/aws-kms": "0.0.0",
     "@aws-cdk/aws-lambda": "0.0.0",
+    "@aws-cdk/aws-logs": "0.0.0",
     "@aws-cdk/aws-s3": "0.0.0",
     "@aws-cdk/aws-sns": "0.0.0",
     "@aws-cdk/aws-sqs": "0.0.0",
     "@aws-cdk/aws-stepfunctions": "0.0.0",
     "@aws-cdk/core": "0.0.0",
+    "@aws-cdk/custom-resources": "0.0.0",
+    "@aws-cdk/lambda-layer-awscli": "0.0.0",
     "constructs": "^3.3.69"
   },
   "homepage": "https://github.com/aws/aws-cdk",
@@ -129,11 +132,14 @@
     "@aws-cdk/aws-iam": "0.0.0",
     "@aws-cdk/aws-kms": "0.0.0",
     "@aws-cdk/aws-lambda": "0.0.0",
+    "@aws-cdk/aws-logs": "0.0.0",
     "@aws-cdk/aws-s3": "0.0.0",
     "@aws-cdk/aws-sns": "0.0.0",
     "@aws-cdk/aws-sqs": "0.0.0",
     "@aws-cdk/aws-stepfunctions": "0.0.0",
     "@aws-cdk/core": "0.0.0",
+    "@aws-cdk/custom-resources": "0.0.0",
+    "@aws-cdk/lambda-layer-awscli": "0.0.0",
     "constructs": "^3.3.69"
   },
   "engines": {

packages/@aws-cdk/aws-stepfunctions-tasks/package.json

@@ -0,0 +1,215 @@
+import { Template } from '@aws-cdk/assertions';
+import * as eks from '@aws-cdk/aws-eks';
+import * as sfn from '@aws-cdk/aws-stepfunctions';
+import { Stack } from '@aws-cdk/core';
+import { EmrContainersCreateVirtualCluster, EksClusterInput } from '../../lib/emrcontainers/create-virtual-cluster';
+
+const emrContainersVirtualClusterName = 'EMR Containers Virtual Cluster';
+let stack: Stack;
+let clusterId: string;
+
+beforeEach(() => {
+  stack = new Stack();
+  clusterId = 'test-eks';
+});
+
+describe('Invoke emr-containers CreateVirtualCluster with ', () => {
+  test('only required properties', () => {
+    // WHEN
+    const task = new EmrContainersCreateVirtualCluster(stack, 'Task', {
+      eksCluster: EksClusterInput.fromTaskInput(sfn.TaskInput.fromText(clusterId)),
+    });
+
+    new sfn.StateMachine(stack, 'SM', {
+      definition: task,
+    });
+
+    // THEN
+    expect(stack.resolve(task.toStateJson())).toEqual({
+      Type: 'Task',
+      Resource: {
+        'Fn::Join': [
+          '',
+          [
+            'arn:',
+            {
+              Ref: 'AWS::Partition',
+            },
+            ':states:::emr-containers:createVirtualCluster',
+          ],
+        ],
+      },
+      End: true,
+      Parameters: {
+        'Name.$': "States.Format('{}/{}', $$.Execution.Name, $$.State.Name)",
+        'ContainerProvider': {
+          Id: clusterId,
+          Info: {
+            EksInfo: {
+              Namespace: 'default',
+            },
+          },
+          Type: 'EKS',
+        },
+      },
+    });
+  });
+
+  test('all required/non-required properties', () => {
+    // WHEN
+    const task = new EmrContainersCreateVirtualCluster(stack, 'Task', {
+      virtualClusterName: emrContainersVirtualClusterName,
+      eksCluster: EksClusterInput.fromTaskInput(sfn.TaskInput.fromText(clusterId)),
+      eksNamespace: 'namespace',
+      integrationPattern: sfn.IntegrationPattern.REQUEST_RESPONSE,
+    });
+
+    new sfn.StateMachine(stack, 'SM', {
+      definition: task,
+    });
+
+    // THEN
+    expect(stack.resolve(task.toStateJson())).toMatchObject({
+      Parameters: {
+        Name: emrContainersVirtualClusterName,
+        ContainerProvider: {
+          Id: clusterId,
+          Info: {
+            EksInfo: {
+              Namespace: 'namespace',
+            },
+          },
+        },
+      },
+    });
+  });
+
+  test('clusterId from payload', () => {
+    // WHEN
+    const task = new EmrContainersCreateVirtualCluster(stack, 'Task', {
+      eksCluster: EksClusterInput.fromTaskInput(sfn.TaskInput.fromJsonPathAt('$.ClusterId')),
+    });
+
+    // THEN
+    expect(stack.resolve(task.toStateJson())).toMatchObject({
+      Parameters: {
+        ContainerProvider: {
+          'Id.$': '$.ClusterId',
+        },
+      },
+    });
+  });
+
+  test('with an existing EKS cluster', () => {
+    // WHEN
+    const eksCluster = new eks.Cluster(stack, 'EKS Cluster', {
+      version: eks.KubernetesVersion.V1_20,
+    });
+
+    const task = new EmrContainersCreateVirtualCluster(stack, 'Task', {
+      eksCluster: EksClusterInput.fromCluster(eksCluster),
+    });
+
+    // THEN
+    expect(stack.resolve(task.toStateJson())).toMatchObject({
+      Parameters: {
+        ContainerProvider: {
+          Id: {
+            Ref: 'EKSClusterEDAD5FD1',
+          },
+        },
+      },
+    });
+  });
+
+  test('Tags', () => {
+    // WHEN
+    const task = new EmrContainersCreateVirtualCluster(stack, 'Task', {
+      eksCluster: EksClusterInput.fromTaskInput(sfn.TaskInput.fromText(clusterId)),
+      tags: {
+        key: 'value',
+      },
+    });
+
+    new sfn.StateMachine(stack, 'SM', {
+      definition: task,
+    });
+
+    // THEN
+    expect(stack.resolve(task.toStateJson())).toMatchObject({
+      Parameters: {
+        Tags: {
+          key: 'value',
+        },
+      },
+    });
+  });
+});
+
+test('Permitted role actions included for CreateVirtualCluster if service integration pattern is REQUEST_RESPONSE', () => {
+  // WHEN
+  const task = new EmrContainersCreateVirtualCluster(stack, 'Task', {
+    virtualClusterName: emrContainersVirtualClusterName,
+    eksCluster: EksClusterInput.fromTaskInput(sfn.TaskInput.fromText(clusterId)),
+  });
+
+  new sfn.StateMachine(stack, 'SM', {
+    definition: task,
+  });
+
+  // THEN
+  Template.fromStack(stack).hasResourceProperties('AWS::IAM::Policy', {
+    PolicyDocument: {
+      Statement: [{
+        Action: 'emr-containers:CreateVirtualCluster',
+        Effect: 'Allow',
+        Resource: '*',
+      },
+      {
+        Action: 'iam:CreateServiceLinkedRole',
+        Condition: {
+          StringLike: {
+            'iam:AWSServiceName': 'emr-containers.amazonaws.com',
+          },
+        },
+        Effect: 'Allow',
+        Resource: {
+          'Fn::Join': [
+            '',
+            [
+              'arn:',
+              {
+                Ref: 'AWS::Partition',
+              },
+              ':iam::',
+              {
+                Ref: 'AWS::AccountId',
+              },
+              ':role/aws-service-role/emr-containers.amazonaws.com/AWSServiceRoleForAmazonEMRContainers',
+            ],
+          ],
+        },
+      }],
+    },
+  });
+});
+
+test('Task throws if WAIT_FOR_TASK_TOKEN is supplied as service integration pattern', () => {
+  expect(() => {
+    new EmrContainersCreateVirtualCluster(stack, 'EMR Containers CreateVirtualCluster Job', {
+      virtualClusterName: emrContainersVirtualClusterName,
+      eksCluster: EksClusterInput.fromTaskInput(sfn.TaskInput.fromText(clusterId)),
+      integrationPattern: sfn.IntegrationPattern.WAIT_FOR_TASK_TOKEN,
+    });
+  }).toThrow(/Unsupported service integration pattern. Supported Patterns: REQUEST_RESPONSE. Received: WAIT_FOR_TASK_TOKEN/);
+});
+
+test('Task throws if RUN_JOB is supplied as service integration pattern', () => {
+  expect(() => {
+    new EmrContainersCreateVirtualCluster(stack, 'EMR Containers CreateVirtualCluster Job', {
+      virtualClusterName: emrContainersVirtualClusterName,
+      eksCluster: EksClusterInput.fromTaskInput(sfn.TaskInput.fromText(clusterId)),
+      integrationPattern: sfn.IntegrationPattern.RUN_JOB,
+    });
+  }).toThrow(/Unsupported service integration pattern. Supported Patterns: REQUEST_RESPONSE. Received: RUN_JOB/);
+});

packages/@aws-cdk/aws-stepfunctions-tasks/test/emrcontainers/create-virtual-cluster.test.ts

@@ -0,0 +1,228 @@
+import { Template } from '@aws-cdk/assertions';
+import * as sfn from '@aws-cdk/aws-stepfunctions';
+import { Stack } from '@aws-cdk/core';
+import { EmrContainersDeleteVirtualCluster } from '../../lib/emrcontainers/delete-virtual-cluster';
+
+let stack: Stack;
+let virtualClusterId: string;
+
+beforeEach(() => {
+  stack = new Stack();
+  virtualClusterId = 'x01f27i9a7cv1td52keaktr6j';
+});
+
+describe('Invoke EMR Containers Delete Virtual cluster with ', () => {
+  test('a valid cluster ID', () => {
+    // WHEN
+    const task = new EmrContainersDeleteVirtualCluster(stack, 'Task', {
+      virtualClusterId: sfn.TaskInput.fromText(virtualClusterId),
+    });
+
+    // THEN
+    expect(stack.resolve(task.toStateJson())).toEqual({
+      Type: 'Task',
+      Resource: {
+        'Fn::Join': [
+          '',
+          [
+            'arn:',
+            {
+              Ref: 'AWS::Partition',
+            },
+            ':states:::emr-containers:deleteVirtualCluster',
+          ],
+        ],
+      },
+      End: true,
+      Parameters: {
+        Id: virtualClusterId,
+      },
+    });
+  });
+
+  test('a RUN_JOB call', () => {
+    // WHEN
+    const task = new EmrContainersDeleteVirtualCluster(stack, 'Task', {
+      virtualClusterId: sfn.TaskInput.fromText(virtualClusterId),
+      integrationPattern: sfn.IntegrationPattern.RUN_JOB,
+    });
+
+    // THEN
+    expect(stack.resolve(task.toStateJson())).toEqual({
+      Type: 'Task',
+      Resource: {
+        'Fn::Join': [
+          '',
+          [
+            'arn:',
+            {
+              Ref: 'AWS::Partition',
+            },
+            ':states:::emr-containers:deleteVirtualCluster.sync',
+          ],
+        ],
+      },
+      End: true,
+      Parameters: {
+        Id: virtualClusterId,
+      },
+    });
+  });
+
+  test('passing in JSON Path', () => {
+    // WHEN
+    const task = new EmrContainersDeleteVirtualCluster(stack, 'Task', {
+      virtualClusterId: sfn.TaskInput.fromJsonPathAt('$.VirtualClusterId'),
+    });
+
+    // THEN
+    expect(stack.resolve(task.toStateJson())).toMatchObject({
+      Parameters: {
+        'Id.$': '$.VirtualClusterId',
+      },
+    });
+  });
+});
+
+describe('Valid policy statements and resources are passed ', () => {
+  test('to the state machine with a REQUEST_RESPONSE call', () => {
+    // WHEN
+    const task = new EmrContainersDeleteVirtualCluster(stack, 'Task', {
+      virtualClusterId: sfn.TaskInput.fromText(virtualClusterId),
+    });
+
+    new sfn.StateMachine(stack, 'SM', {
+      definition: task,
+    });
+
+    // THEN
+    Template.fromStack(stack).hasResourceProperties('AWS::IAM::Policy', {
+      PolicyDocument: {
+        Statement: [{
+          Action: 'emr-containers:DeleteVirtualCluster',
+          Effect: 'Allow',
+          Resource: {
+            'Fn::Join': [
+              '',
+              [
+                'arn:',
+                {
+                  Ref: 'AWS::Partition',
+                },
+                ':emr-containers:',
+                {
+                  Ref: 'AWS::Region',
+                },
+                ':',
+                {
+                  Ref: 'AWS::AccountId',
+                },
+                `:/virtualclusters/${virtualClusterId}`,
+              ],
+            ],
+          },
+        }],
+      },
+    });
+  });
+
+  test('to the state machine with a RUN_JOB call', () => {
+    // WHEN
+    const task = new EmrContainersDeleteVirtualCluster(stack, 'Task', {
+      virtualClusterId: sfn.TaskInput.fromText(virtualClusterId),
+      integrationPattern: sfn.IntegrationPattern.RUN_JOB,
+    });
+
+    new sfn.StateMachine(stack, 'SM', {
+      definition: task,
+    });
+
+    // THEN
+    Template.fromStack(stack).hasResourceProperties('AWS::IAM::Policy', {
+      PolicyDocument: {
+        Statement: [{
+          Action: [
+            'emr-containers:DeleteVirtualCluster',
+            'emr-containers:DescribeVirtualCluster',
+          ],
+          Effect: 'Allow',
+          Resource: {
+            'Fn::Join': [
+              '',
+              [
+                'arn:',
+                {
+                  Ref: 'AWS::Partition',
+                },
+                ':emr-containers:',
+                {
+                  Ref: 'AWS::Region',
+                },
+                ':',
+                {
+                  Ref: 'AWS::AccountId',
+                },
+                `:/virtualclusters/${virtualClusterId}`,
+              ],
+            ],
+          },
+        }],
+      },
+    });
+  });
+
+  test('when the virtual cluster ID is from a payload', () => {
+    // WHEN
+    const task = new EmrContainersDeleteVirtualCluster(stack, 'Task', {
+      virtualClusterId: sfn.TaskInput.fromJsonPathAt('$.ClusterId'),
+      integrationPattern: sfn.IntegrationPattern.RUN_JOB,
+    });
+
+    new sfn.StateMachine(stack, 'SM', {
+      definition: task,
+    });
+
+    // THEN
+    Template.fromStack(stack).hasResourceProperties('AWS::IAM::Policy', {
+      PolicyDocument: {
+        Statement: [{
+          Action: [
+            'emr-containers:DeleteVirtualCluster',
+            'emr-containers:DescribeVirtualCluster',
+          ],
+          Effect: 'Allow',
+          Resource: {
+            'Fn::Join': [
+              '',
+              [
+                'arn:',
+                {
+                  Ref: 'AWS::Partition',
+                },
+                ':emr-containers:',
+                {
+                  Ref: 'AWS::Region',
+                },
+                ':',
+                {
+                  Ref: 'AWS::AccountId',
+                },
+                ':/virtualclusters/*',
+              ],
+            ],
+          },
+        }],
+      },
+    });
+  });
+});
+
+
+test('Task throws if WAIT_FOR_TASK_TOKEN is supplied as service integration pattern', () => {
+  expect(() => {
+    new EmrContainersDeleteVirtualCluster(stack, 'EMR Containers DeleteVirtualCluster Job', {
+      virtualClusterId: sfn.TaskInput.fromText(virtualClusterId),
+      integrationPattern: sfn.IntegrationPattern.WAIT_FOR_TASK_TOKEN,
+    });
+  }).toThrow(/Unsupported service integration pattern. Supported Patterns: REQUEST_RESPONSE,RUN_JOB. Received: WAIT_FOR_TASK_TOKEN/);
+});

packages/@aws-cdk/aws-stepfunctions-tasks/test/emrcontainers/delete-virtual-cluster.test.ts

@@ -0,0 +1,90 @@
+import * as ec2 from '@aws-cdk/aws-ec2';
+import * as eks from '@aws-cdk/aws-eks';
+import * as iam from '@aws-cdk/aws-iam';
+import * as sfn from '@aws-cdk/aws-stepfunctions';
+import * as cdk from '@aws-cdk/core';
+import {
+  Classification, VirtualClusterInput, EksClusterInput, EmrContainersDeleteVirtualCluster,
+  EmrContainersCreateVirtualCluster, EmrContainersStartJobRun, ReleaseLabel,
+} from '../../lib';
+
+/**
+ * Stack verification steps:
+ * Everything in the links below must be setup for the EKS Cluster and Execution Role before running the state machine.
+ * @see https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/setting-up-cluster-access.html
+ * @see https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/setting-up-enable-IAM.html
+ * @see https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/setting-up-trust-policy.html
+ *
+ * aws stepfunctions start-execution --state-machine-arn <deployed state machine arn> : should return execution arn
+ * aws stepfunctions describe-execution --execution-arn <exection-arn generated before> : should return status as SUCCEEDED
+ */
+
+const app = new cdk.App();
+const stack = new cdk.Stack(app, 'aws-stepfunctions-tasks-emr-containers-all-services-integ');
+
+const eksCluster = new eks.Cluster(stack, 'integration-test-eks-cluster', {
+  version: eks.KubernetesVersion.V1_21,
+  defaultCapacity: 3,
+  defaultCapacityInstance: ec2.InstanceType.of(ec2.InstanceClass.M5, ec2.InstanceSize.XLARGE),
+});
+
+const jobExecutionRole = new iam.Role(stack, 'JobExecutionRole', {
+  assumedBy: new iam.CompositePrincipal(
+    new iam.ServicePrincipal('emr-containers.amazonaws.com'),
+    new iam.ServicePrincipal('states.amazonaws.com'),
+  ),
+});
+
+const createVirtualCluster = new EmrContainersCreateVirtualCluster(stack, 'Create a virtual Cluster', {
+  virtualClusterName: 'Virtual-Cluster-Name',
+  eksCluster: EksClusterInput.fromCluster(eksCluster),
+  resultPath: '$.cluster',
+});
+
+const startJobRun = new EmrContainersStartJobRun(stack, 'Start a Job Run', {
+  virtualCluster: VirtualClusterInput.fromTaskInput(sfn.TaskInput.fromJsonPathAt('$.cluster.Id')),
+  releaseLabel: ReleaseLabel.EMR_6_2_0,
+  jobName: 'EMR-Containers-Job',
+  executionRole: iam.Role.fromRoleArn(stack, 'Job-Execution-Role', jobExecutionRole.roleArn),
+  jobDriver: {
+    sparkSubmitJobDriver: {
+      entryPoint: sfn.TaskInput.fromText('local:///usr/lib/spark/examples/src/main/python/pi.py'),
+      entryPointArguments: sfn.TaskInput.fromObject(['2']),
+      sparkSubmitParameters: '--conf spark.driver.memory=512M --conf spark.kubernetes.driver.request.cores=0.2 --conf spark.kubernetes.executor.request.cores=0.2 --conf spark.sql.shuffle.partitions=60 --conf spark.dynamicAllocation.enabled=false',
+    },
+  },
+  monitoring: {
+    logging: true,
+    persistentAppUI: true,
+  },
+  applicationConfig: [{
+    classification: Classification.SPARK_DEFAULTS,
+    properties: {
+      'spark.executor.instances': '1',
+      'spark.executor.memory': '512M',
+    },
+  }],
+  resultPath: '$.job',
+});
+
+
+const deleteVirtualCluster = new EmrContainersDeleteVirtualCluster(stack, 'Delete a Virtual Cluster', {
+  virtualClusterId: sfn.TaskInput.fromJsonPathAt('$.job.VirtualClusterId'),
+});
+
+const chain = sfn.Chain
+  .start(createVirtualCluster)
+  .next(startJobRun)
+  .next(deleteVirtualCluster);
+
+const sm = new sfn.StateMachine(stack, 'StateMachine', {
+  definition: chain,
+  timeout: cdk.Duration.minutes(20),
+});
+
+new cdk.CfnOutput(stack, 'stateMachineArn', {
+  value: sm.stateMachineArn,
+});
+
+
+app.synth();

packages/@aws-cdk/aws-stepfunctions-tasks/test/emrcontainers/integ.job-submission-workflow.expected.json

@@ -0,0 +1,100 @@
+import * as ec2 from '@aws-cdk/aws-ec2';
+import * as eks from '@aws-cdk/aws-eks';
+import { AwsAuthMapping } from '@aws-cdk/aws-eks';
+import * as iam from '@aws-cdk/aws-iam';
+import * as sfn from '@aws-cdk/aws-stepfunctions';
+import * as cdk from '@aws-cdk/core';
+import { Aws } from '@aws-cdk/core';
+import { EmrContainersStartJobRun } from '../../lib';
+import { ReleaseLabel, VirtualClusterInput } from '../../lib/emrcontainers/start-job-run';
+
+/**
+ * Stack verification steps:
+ * Everything in the link below must be setup before running the state machine.
+ * @see https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/setting-up-enable-IAM.html
+ * aws stepfunctions start-execution --state-machine-arn <deployed state machine arn> : should return execution arn
+ * aws stepfunctions describe-execution --execution-arn <exection-arn generated before> : should return status as SUCCEEDED
+ */
+
+const app = new cdk.App();
+const stack = new cdk.Stack(app, 'aws-stepfunctions-tasks-emr-containers-start-job-run-integ-test');
+
+const eksCluster = new eks.Cluster(stack, 'integration-test-eks-cluster', {
+  version: eks.KubernetesVersion.V1_21,
+  defaultCapacity: 3,
+  defaultCapacityInstance: ec2.InstanceType.of(ec2.InstanceClass.M5, ec2.InstanceSize.XLARGE),
+});
+
+const virtualCluster = new cdk.CfnResource(stack, 'Virtual Cluster', {
+  type: 'AWS::EMRContainers::VirtualCluster',
+  properties: {
+    ContainerProvider: {
+      Id: eksCluster.clusterName,
+      Info: {
+        EksInfo: {
+          Namespace: 'default',
+        },
+      },
+      Type: 'EKS',
+    },
+    Name: 'Virtual-Cluster-Name',
+  },
+});
+
+const emrRole = eksCluster.addManifest('emrRole', {
+  apiVersion: 'rbac.authorization.k8s.io/v1',
+  kind: 'Role',
+  metadata: { name: 'emr-containers', namespace: 'default' },
+  rules: [
+    { apiGroups: [''], resources: ['namespaces'], verbs: ['get'] },
+    { apiGroups: [''], resources: ['serviceaccounts', 'services', 'configmaps', 'events', 'pods', 'pods/log'], verbs: ['get', 'list', 'watch', 'describe', 'create', 'edit', 'delete', 'deletecollection', 'annotate', 'patch', 'label'] },
+    { apiGroups: [''], resources: ['secrets'], verbs: ['create', 'patch', 'delete', 'watch'] },
+    { apiGroups: ['apps'], resources: ['statefulsets', 'deployments'], verbs: ['get', 'list', 'watch', 'describe', 'create', 'edit', 'delete', 'annotate', 'patch', 'label'] },
+    { apiGroups: ['batch'], resources: ['jobs'], verbs: ['get', 'list', 'watch', 'describe', 'create', 'edit', 'delete', 'annotate', 'patch', 'label'] },
+    { apiGroups: ['extensions'], resources: ['ingresses'], verbs: ['get', 'list', 'watch', 'describe', 'create', 'edit', 'delete', 'annotate', 'patch', 'label'] },
+    { apiGroups: ['rbac.authorization.k8s.io'], resources: ['roles', 'rolebindings'], verbs: ['get', 'list', 'watch', 'describe', 'create', 'edit', 'delete', 'deletecollection', 'annotate', 'patch', 'label'] },
+  ],
+});
+
+const emrRoleBind = eksCluster.addManifest('emrRoleBind', {
+  apiVersion: 'rbac.authorization.k8s.io/v1',
+  kind: 'RoleBinding',
+  metadata: { name: 'emr-containers', namespace: 'default' },
+  subjects: [{ kind: 'User', name: 'emr-containers', apiGroup: 'rbac.authorization.k8s.io' }],
+  roleRef: { kind: 'Role', name: 'emr-containers', apiGroup: 'rbac.authorization.k8s.io' },
+});
+
+emrRoleBind.node.addDependency(emrRole);
+
+const emrServiceRole = iam.Role.fromRoleArn(stack, 'emrServiceRole', 'arn:aws:iam::'+Aws.ACCOUNT_ID+':role/AWSServiceRoleForAmazonEMRContainers');
+const authMapping: AwsAuthMapping = { groups: [], username: 'emr-containers' };
+eksCluster.awsAuth.addRoleMapping(emrServiceRole, authMapping);
+
+virtualCluster.node.addDependency(emrRoleBind);
+virtualCluster.node.addDependency(eksCluster.awsAuth);
+
+const startJobRunJob = new EmrContainersStartJobRun(stack, 'Start a Job Run', {
+  virtualCluster: VirtualClusterInput.fromVirtualClusterId(virtualCluster.getAtt('Id').toString()),
+  releaseLabel: ReleaseLabel.EMR_6_2_0,
+  jobName: 'EMR-Containers-Job',
+  jobDriver: {
+    sparkSubmitJobDriver: {
+      entryPoint: sfn.TaskInput.fromText('local:///usr/lib/spark/examples/src/main/python/pi.py'),
+      entryPointArguments: sfn.TaskInput.fromObject(['2']),
+      sparkSubmitParameters: '--conf spark.driver.memory=512M --conf spark.kubernetes.driver.request.cores=0.2 --conf spark.kubernetes.executor.request.cores=0.2 --conf spark.sql.shuffle.partitions=60 --conf spark.dynamicAllocation.enabled=false',
+    },
+  },
+});
+
+const chain = sfn.Chain.start(startJobRunJob);
+
+const sm = new sfn.StateMachine(stack, 'StateMachine', {
+  definition: chain,
+  timeout: cdk.Duration.seconds(1000),
+});
+
+new cdk.CfnOutput(stack, 'stateMachineArn', {
+  value: sm.stateMachineArn,
+});
+
+app.synth();