feat(msk): support ServerlessCluster

packages/@aws-cdk/aws-msk-alpha/README.md

@@ -231,3 +231,24 @@ const cluster = new msk.Cluster(this, 'cluster', {
   storageMode: msk.StorageMode.TIERED,
 });
 ```
+
+## MSK Serverless
+
+You can also use MSK Serverless by using `ServerlessCluster` class.
+
+MSK Serverless is a cluster type for Amazon MSK that makes it possible for you to run Apache Kafka without having to manage and scale cluster capacity.
+
+MSK Serverless requires IAM access control for all clusters.
+
+For more infomation, see [Use MSK Serverless clusters](https://docs.aws.amazon.com/msk/latest/developerguide/serverless-getting-started.html).
+
+```ts
+declare const vpc: ec2.Vpc;
+
+const serverlessCluster = new msk.ServerlessCluster(this, 'ServerlessCluster', {
+  clusterName: 'MyServerlessCluster',
+  vpcConfigs: [
+    { vpc },
+  ],
+});
+```

packages/@aws-cdk/aws-msk-alpha/awslint.json

@@ -1,4 +1,5 @@
 {
     "exclude": [
+        "props-physical-name:@aws-cdk/aws-msk-alpha.ServerlessClusterProps"
     ]
 }

packages/@aws-cdk/aws-msk-alpha/lib/cluster.ts

@@ -36,7 +36,7 @@ export interface ICluster extends core.IResource, ec2.IConnectable {
 /**
  * A new or imported MSK Cluster.
  */
-abstract class ClusterBase extends core.Resource implements ICluster {
+export abstract class ClusterBase extends core.Resource implements ICluster {
   public abstract readonly clusterArn: string;
   public abstract readonly clusterName: string;
   /** @internal */

packages/@aws-cdk/aws-msk-alpha/lib/index.ts

@@ -1,4 +1,5 @@
 export * from './cluster';
 export * from './cluster-version';
+export * from './serverless-cluster';
 
 // AWS::MSK CloudFormation Resources:

packages/@aws-cdk/aws-msk-alpha/lib/serverless-cluster.ts

@@ -0,0 +1,147 @@
+import * as ec2 from 'aws-cdk-lib/aws-ec2';
+import { Fn, Lazy, Names } from 'aws-cdk-lib';
+import * as constructs from 'constructs';
+import { ClusterBase, ICluster } from '.';
+import { CfnServerlessCluster } from 'aws-cdk-lib/aws-msk';
+
+/**
+ *  Properties for a MSK Serverless Cluster
+ */
+export interface ServerlessClusterProps {
+  /**
+   * The physical name of the cluster.
+   *
+   * @default - auto generate
+   */
+  readonly clusterName?: string;
+
+  /**
+   * The configuration of the Amazon VPCs for the cluster.
+   * You can specify up to 5 VPC configurations.
+   */
+  readonly vpcConfigs: VpcConfig[];
+}
+
+/**
+ * The configuration of the Amazon VPCs for the cluster.
+ */
+export interface VpcConfig {
+  /**
+   * Defines the virtual networking environment for this cluster.
+   * Must have at least 2 subnets in two different AZs.
+   */
+  readonly vpc: ec2.IVpc;
+
+  /**
+   * The subnets associated with the cluster.
+   *
+   * @default - the Vpc default strategy if not specified.
+   */
+  readonly vpcSubnets?: ec2.SubnetSelection;
+
+  /**
+   * The security groups associated with the cluster.
+   * You can specify up to 5 security groups.
+   *
+   * @default - create new security group
+   */
+  readonly securityGroups?: ec2.ISecurityGroup[];
+}
+
+/**
+ * Create a MSK Serverless Cluster.
+ *
+ * @resource AWS::MSK::ServerlessCluster
+ */
+export class ServerlessCluster extends ClusterBase {
+  /**
+   * Reference an existing cluster, defined outside of the CDK code, by name.
+   */
+  public static fromClusterArn(
+    scope: constructs.Construct,
+    id: string,
+    clusterArn: string,
+  ): ICluster {
+    class Import extends ClusterBase {
+      public readonly clusterArn = clusterArn;
+      public readonly clusterName = Fn.select(1, Fn.split('/', clusterArn)); // ['arn:partition:kafka:region:account-id', clusterName, clusterId]
+    }
+
+    return new Import(scope, id);
+  }
+
+  public readonly clusterArn: string;
+  public readonly clusterName: string;
+
+  private _securityGroups: ec2.ISecurityGroup[] = [];
+
+  constructor(scope: constructs.Construct, id: string, props: ServerlessClusterProps) {
+    super(scope, id, {
+      physicalName: props.clusterName ??
+        Lazy.string({
+          produce: () => Names.uniqueResourceName(this, { maxLength: 64 }),
+        }),
+    });
+
+    if (props.vpcConfigs.length < 1 || props.vpcConfigs.length > 5) {
+      throw Error(`\`vpcConfigs\` must contain between 1 and 5 configurations, got ${props.vpcConfigs.length} configurations.`);
+    }
+
+    const vpcConfigs = props.vpcConfigs.map((vpcConfig, index) => this._renderVpcConfig(vpcConfig, index));
+
+    this._connections = new ec2.Connections({
+      securityGroups: this._securityGroups,
+    });
+
+    const resource = new CfnServerlessCluster(this, 'Resource', {
+      clusterName: this.physicalName,
+      clientAuthentication: {
+        sasl: {
+          iam: {
+            enabled: true,
+          },
+        },
+      },
+      vpcConfigs,
+    });
+
+    this.clusterName = this.getResourceNameAttribute(
+      Fn.select(1, Fn.split('/', resource.ref)),
+    );
+    this.clusterArn = resource.ref;
+  }
+
+  /**
+   * Render Vpc Config property
+   */
+  private _renderVpcConfig(vpcConfig: VpcConfig, index: number): CfnServerlessCluster.VpcConfigProperty {
+    const subnetSelection = vpcConfig.vpc.selectSubnets(vpcConfig.vpcSubnets);
+
+    if (subnetSelection.subnets.length < 2) {
+      throw Error(
+        `Cluster requires at least 2 subnets, got ${subnetSelection.subnets.length}.`,
+      );
+    }
+
+    let securityGroups: ec2.ISecurityGroup[] = [];
+
+    if (vpcConfig.securityGroups) {
+      if (vpcConfig.securityGroups.length < 1 || vpcConfig.securityGroups.length > 5) {
+        throw Error(`\`securityGroups\` must contain between 1 and 5 elements, got ${vpcConfig.securityGroups.length} elements.`);
+      }
+      securityGroups = vpcConfig.securityGroups;
+    } else {
+      securityGroups.push(new ec2.SecurityGroup(this, `SecurityGroup-${index}`, {
+        description: 'MSK Serverless security group',
+        vpc: vpcConfig.vpc,
+      }));
+    }
+
+    this._securityGroups.push(...securityGroups);
+
+    return {
+      subnetIds: subnetSelection.subnets.map((subnet) => subnet.subnetId),
+      securityGroups: securityGroups?.map((securityGroup) => securityGroup.securityGroupId),
+    };
+  }
+}