feat(msk): Cluster L2 Construct (#9908)

L2 Construct for a MSK Cluster. 

I wrote this for internal use and thought I'd share it. I tried to follow the [example resource](https://github.com/aws/aws-cdk/blob/master/packages/%40aws-cdk/example-construct-library/lib/example-resource.ts) and [design guidelines](https://github.com/aws/aws-cdk/blob/master/DESIGN_GUIDELINES.md) as much as I could. Default properties were chosen either based on defaults when creating a cluster in the console or defaults set from CloudFormation.

Closes #9603

----

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

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

@@ -21,8 +21,96 @@
 
 <!--END STABILITY BANNER-->
 
-This module is part of the [AWS Cloud Development Kit](https://github.com/aws/aws-cdk) project.
+[Amazon MSK](https://aws.amazon.com/msk/) is a fully managed service that makes it easy for you to build and run applications that use Apache Kafka to process streaming data.
+
+The following example creates an MSK Cluster.
 
 ```ts
 import * as msk from '@aws-cdk/aws-msk';
+
+const cluster = new Cluster(this, 'Cluster', {
+  kafkaVersion: msk.KafkaVersion.V2_6_1,
+  vpc,
+});
+```
+
+## Allowing Connections
+
+To control who can access the Cluster, use the `.connections` attribute. For a list of ports used by MSK, refer to the [MSK documentation](https://docs.aws.amazon.com/msk/latest/developerguide/client-access.html#port-info).
+
+```typescript
+import * as msk from "@aws-cdk/aws-msk"
+import * as ec2 from "@aws-cdk/aws-ec2"
+
+const cluster = new msk.Cluster(this, "Cluster", {...})
+
+cluster.connections.allowFrom(
+  ec2.Peer.ipv4("1.2.3.4/8"),
+  ec2.Port.tcp(2181)
+)
+cluster.connections.allowFrom(
+  ec2.Peer.ipv4("1.2.3.4/8"),
+  ec2.Port.tcp(9094)
+)
+```
+
+## Cluster Endpoints
+
+You can use the following attributes to get a list of the Kafka broker or ZooKeeper node endpoints
+
+```typescript
+new cdk.CfnOutput(this, 'BootstrapBrokers', { value: cluster.bootstrapBrokers });
+new cdk.CfnOutput(this, 'BootstrapBrokersTls', { value: cluster.bootstrapBrokersTls });
+new cdk.CfnOutput(this, 'BootstrapBrokersSaslScram', { value: cluster.bootstrapBrokersSaslScram });
+new cdk.CfnOutput(this, 'ZookeeperConnection', { value: cluster.zookeeperConnectionString });
+new cdk.CfnOutput(this, 'ZookeeperConnectionTls', { value: cluster.zookeeperConnectionStringTls });
+```
+
+## Importing an existing Cluster
+
+To import an existing MSK cluster into your CDK app use the `.fromClusterArn()` method.
+
+```typescript
+const cluster = msk.Cluster.fromClusterArn(this, 'Cluster', 'arn:aws:kafka:us-west-2:1234567890:cluster/a-cluster/11111111-1111-1111-1111-111111111111-1')
+```
+
+## Client Authentication
+
+### TLS
+
+To enable client authentication with TLS set the `certificateAuthorityArns` property to reference your ACM Private CA. [More info on Private CAs.](https://docs.aws.amazon.com/msk/latest/developerguide/msk-authentication.html)
+
+```typescript
+import * as msk from "@aws-cdk/aws-msk"
+
+const cluster = new msk.Cluster(this, 'Cluster', {
+    ...
+    encryptionInTransit: {
+      clientBroker: msk.ClientBrokerEncryption.TLS,
+    },
+    clientAuthentication: msk.ClientAuthentication.tls({
+      certificateAuthorityArns: [
+        'arn:aws:acm-pca:us-west-2:1234567890:certificate-authority/11111111-1111-1111-1111-111111111111',
+      ],
+    }),
+  });
+});
+```
+
+### SASL/SCRAM
+
+Enable client authentication with SASL/SCRAM:
+
+```typescript
+import * as msk from "@aws-cdk/aws-msk"
+
+const cluster = new msk.cluster(this, "cluster", {
+  ...
+  encryptionInTransit: {
+    clientBroker: msk.ClientBrokerEncryption.TLS,
+  },
+  clientAuthentication: msk.ClientAuthentication.sasl({
+    scram: true,
+  }),
+})
 ```

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

@@ -0,0 +1,63 @@
+/**
+ * Kafka cluster version
+ */
+export class KafkaVersion {
+  /**
+   * Kafka version 1.1.1
+   */
+  public static readonly V1_1_1 = KafkaVersion.of('1.1.1');
+
+  /**
+   * Kafka version 2.2.1
+   */
+  public static readonly V2_2_1 = KafkaVersion.of('2.2.1');
+
+  /**
+   * Kafka version 2.3.1
+   */
+  public static readonly V2_3_1 = KafkaVersion.of('2.3.1');
+
+  /**
+   * Kafka version 2.4.1
+   */
+  public static readonly V2_4_1_1 = KafkaVersion.of('2.4.1.1');
+
+  /**
+   * Kafka version 2.5.1
+   */
+  public static readonly V2_5_1 = KafkaVersion.of('2.5.1');
+
+  /**
+   * Kafka version 2.6.0
+   */
+  public static readonly V2_6_0 = KafkaVersion.of('2.6.0');
+
+  /**
+   * Kafka version 2.6.1
+   */
+  public static readonly V2_6_1 = KafkaVersion.of('2.6.1');
+
+  /**
+   * Kafka version 2.7.0
+   */
+  public static readonly V2_7_0 = KafkaVersion.of('2.7.0');
+
+  /**
+   * Kafka version 2.8.0
+   */
+  public static readonly V2_8_0 = KafkaVersion.of('2.8.0');
+
+  /**
+   * Custom cluster version
+   * @param version custom version number
+   */
+  public static of(version: string) {
+    return new KafkaVersion(version);
+  }
+
+  /**
+   *
+   * @param version cluster version number
+   */
+  private constructor(public readonly version: string) {}
+}