feat(msk): support Kafka 4.0 (#34501)

### Issue # (if applicable)

N/A
### Reason for this change
MS support Apache Kafka version 4.0.


### Description of changes
Add ver 4.0 enum.
Kafka version 4.0 only supports KRaft (Apache Kafka Raft).


### Describe any new or updated permissions being added
nothing.



### Description of how you validated changes
Add a unit test and an integ test.


### 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: mazyu36

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

@@ -23,7 +23,7 @@ The following example creates an MSK Cluster.
 declare const vpc: ec2.Vpc;
 const cluster = new msk.Cluster(this, 'Cluster', {
   clusterName: 'myCluster',
-  kafkaVersion: msk.KafkaVersion.V3_9_X,
+  kafkaVersion: msk.KafkaVersion.V4_0_X_KRAFT,
   vpc,
 });
 ```
@@ -36,7 +36,7 @@ To control who can access the Cluster, use the `.connections` attribute. For a l
 declare const vpc: ec2.Vpc;
 const cluster = new msk.Cluster(this, 'Cluster', {
   clusterName: 'myCluster',
-  kafkaVersion: msk.KafkaVersion.V3_9_X,
+  kafkaVersion: msk.KafkaVersion.V4_0_X_KRAFT,
   vpc,
 });
 
@@ -88,7 +88,7 @@ import * as acmpca from 'aws-cdk-lib/aws-acmpca';
 declare const vpc: ec2.Vpc;
 const cluster = new msk.Cluster(this, 'Cluster', {
   clusterName: 'myCluster',
-  kafkaVersion: msk.KafkaVersion.V3_9_X,
+  kafkaVersion: msk.KafkaVersion.V4_0_X_KRAFT,
   vpc,
   encryptionInTransit: {
     clientBroker: msk.ClientBrokerEncryption.TLS,
@@ -113,7 +113,7 @@ Enable client authentication with [SASL/SCRAM](https://docs.aws.amazon.com/msk/l
 declare const vpc: ec2.Vpc;
 const cluster = new msk.Cluster(this, 'cluster', {
   clusterName: 'myCluster',
-  kafkaVersion: msk.KafkaVersion.V3_9_X,
+  kafkaVersion: msk.KafkaVersion.V4_0_X_KRAFT,
   vpc,
   encryptionInTransit: {
     clientBroker: msk.ClientBrokerEncryption.TLS,
@@ -132,7 +132,7 @@ Enable client authentication with [IAM](https://docs.aws.amazon.com/msk/latest/d
 declare const vpc: ec2.Vpc;
 const cluster = new msk.Cluster(this, 'cluster', {
   clusterName: 'myCluster',
-  kafkaVersion: msk.KafkaVersion.V3_9_X,
+  kafkaVersion: msk.KafkaVersion.V4_0_X_KRAFT,
   vpc,
   encryptionInTransit: {
     clientBroker: msk.ClientBrokerEncryption.TLS,
@@ -155,7 +155,7 @@ import * as acmpca from 'aws-cdk-lib/aws-acmpca';
 declare const vpc: ec2.Vpc;
 const cluster = new msk.Cluster(this, 'Cluster', {
   clusterName: 'myCluster',
-  kafkaVersion: msk.KafkaVersion.V3_9_X,
+  kafkaVersion: msk.KafkaVersion.V4_0_X_KRAFT,
   vpc,
   encryptionInTransit: {
     clientBroker: msk.ClientBrokerEncryption.TLS,
@@ -186,7 +186,7 @@ declare const vpc: ec2.Vpc;
 declare const bucket: s3.IBucket;
 const cluster = new msk.Cluster(this, 'cluster', {
   clusterName: 'myCluster',
-  kafkaVersion: msk.KafkaVersion.V3_9_X,
+  kafkaVersion: msk.KafkaVersion.V4_0_X_KRAFT,
   vpc,
   logging: {
     s3: {
@@ -226,7 +226,7 @@ declare const bucket: s3.IBucket;
 
 const cluster = new msk.Cluster(this, 'cluster', {
   clusterName: 'myCluster',
-  kafkaVersion: msk.KafkaVersion.V3_9_X,
+  kafkaVersion: msk.KafkaVersion.V4_0_X_KRAFT,
   vpc,
   storageMode: msk.StorageMode.TIERED,
 });

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

@@ -224,6 +224,14 @@ export class KafkaVersion {
    * @see https://docs.aws.amazon.com/msk/latest/developerguide/metadata-management.html#kraft-intro
    */
   public static readonly V3_9_X_KRAFT = KafkaVersion.of('3.9.x.kraft', { tieredStorage: true });
+
+  /**
+   * Kafka version 4.0.x with KRaft (Apache Kafka Raft) metadata mode support
+   *
+   * @see https://docs.aws.amazon.com/msk/latest/developerguide/metadata-management.html#kraft-intro
+   */
+  public static readonly V4_0_X_KRAFT = KafkaVersion.of('4.0.x.kraft', { tieredStorage: true });
+
   /**
    * Custom cluster version
    * @param version custom version number

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

@@ -51,6 +51,7 @@ describe('MSK Cluster', () => {
     [msk.KafkaVersion.V3_8_X_KRAFT, '3.8.x.kraft'],
     [msk.KafkaVersion.V3_9_X, '3.9.x'],
     [msk.KafkaVersion.V3_9_X_KRAFT, '3.9.x.kraft'],
+    [msk.KafkaVersion.V4_0_X_KRAFT, '4.0.x.kraft'],
   ],
   )('created with expected Kafka version %j', (parameter, result) => {
     new msk.Cluster(stack, 'Cluster', {

packages/@aws-cdk/aws-msk-alpha/test/integ.cluster-version.js.snapshot/KafkaVersionTestStack.assets.json

@@ -1977,6 +1977,75 @@
    },
    "UpdateReplacePolicy": "Delete",
    "DeletionPolicy": "Delete"
+  },
+  "ClusterVersion40xkraftSecurityGroup81D50A3C": {
+   "Type": "AWS::EC2::SecurityGroup",
+   "Properties": {
+    "GroupDescription": "MSK security group",
+    "SecurityGroupEgress": [
+     {
+      "CidrIp": "0.0.0.0/0",
+      "Description": "Allow all outbound traffic by default",
+      "IpProtocol": "-1"
+     }
+    ],
+    "VpcId": {
+     "Ref": "Vpc8378EB38"
+    }
+   }
+  },
+  "ClusterVersion40xkraft5E11CA80": {
+   "Type": "AWS::MSK::Cluster",
+   "Properties": {
+    "BrokerNodeGroupInfo": {
+     "ClientSubnets": [
+      {
+       "Ref": "VpcPrivateSubnet1Subnet536B997A"
+      },
+      {
+       "Ref": "VpcPrivateSubnet2Subnet3788AAA1"
+      }
+     ],
+     "InstanceType": "kafka.m5.large",
+     "SecurityGroups": [
+      {
+       "Fn::GetAtt": [
+        "ClusterVersion40xkraftSecurityGroup81D50A3C",
+        "GroupId"
+       ]
+      }
+     ],
+     "StorageInfo": {
+      "EBSStorageInfo": {
+       "VolumeSize": 1000
+      }
+     }
+    },
+    "ClusterName": "cluster-v4-0-x-kraft",
+    "EncryptionInfo": {
+     "EncryptionInTransit": {
+      "ClientBroker": "TLS",
+      "InCluster": true
+     }
+    },
+    "KafkaVersion": "4.0.x.kraft",
+    "LoggingInfo": {
+     "BrokerLogs": {
+      "CloudWatchLogs": {
+       "Enabled": false
+      },
+      "Firehose": {
+       "Enabled": false
+      },
+      "S3": {
+       "Enabled": false
+      }
+     }
+    },
+    "NumberOfBrokerNodes": 2
+   },
+   "UpdateReplacePolicy": "Delete",
+   "DeletionPolicy": "Delete"
   }
  },
  "Parameters": {

packages/@aws-cdk/aws-msk-alpha/test/integ.cluster-version.js.snapshot/KafkaVersionTestStack.template.json

@@ -29,6 +29,7 @@ const versions: KafkaVersion[] = [
   KafkaVersion.V3_8_X_KRAFT,
   KafkaVersion.V3_9_X,
   KafkaVersion.V3_9_X_KRAFT,
+  KafkaVersion.V4_0_X_KRAFT,
 ];
 
 class KafkaVersionTest extends Stack {