feat(msk-alpha): MSK Kafka versions 2.8.2.tiered and 3.5.1 and StorageMode property

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

@@ -76,6 +76,11 @@ export class KafkaVersion {
    */
   public static readonly V2_8_1 = KafkaVersion.of('2.8.1');
 
+  /**
+   * AWS MSK Kafka version 2.8.2.tiered
+   */
+  public static readonly V2_8_2_TIERED = KafkaVersion.of('2.8.2.tiered');
+
   /**
    * Kafka version 3.1.1
    */
@@ -101,6 +106,11 @@ export class KafkaVersion {
    */
   public static readonly V3_4_0 = KafkaVersion.of('3.4.0');
 
+  /**
+   * Kafka version 3.5.1
+   */
+  public static readonly V3_5_1 = KafkaVersion.of('3.5.1');
+
   /**
    * Custom cluster version
    * @param version custom version number
@@ -109,6 +119,20 @@ export class KafkaVersion {
     return new KafkaVersion(version);
   }
 
+  /**
+   * Does this Kafka version support tiered storage?
+   * @param version cluster kafka version
+   * MSK Kafka versions that support tiered storage.
+   * @see https://docs.aws.amazon.com/msk/latest/developerguide/msk-tiered-storage.html#msk-tiered-storage-requirements
+   */
+  public static isTieredStorageCompatible(version: KafkaVersion | string) {
+    if (version instanceof KafkaVersion) {
+      version = version.version;
+    }
+    const suffix = 'tiered';
+    return version.indexOf(suffix, version.length - suffix.length) !== -1;
+  };
+
   /**
    *
    * @param version cluster version number

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

@@ -103,6 +103,12 @@ export interface ClusterProps {
    * @default - 1000 GiB EBS volume
    */
   readonly ebsStorageInfo?: EbsStorageInfo;
+  /**
+   * Local or Tiered storage configuration for the brokers
+   *
+   * @default - StorageMode.LOCAL
+   */
+  readonly storageMode?: StorageMode;
   /**
    * The Amazon MSK configuration to use for the cluster.
    *
@@ -160,6 +166,22 @@ export interface EbsStorageInfo {
   readonly encryptionKey?: kms.IKey;
 }
 
+/**
+ * This controls storage mode for supported storage tiers.
+ *
+ * @see https://docs.aws.amazon.com/msk/latest/developerguide/msk-tiered-storage.html
+ */
+export enum StorageMode {
+  /**
+   * Local storage mode utilizes network attached EBS storage.
+   */
+  LOCAL = 'LOCAL',
+  /**
+   * Tiered storage mode utilizes EBS storage and S3.
+   */
+  TIERED = 'TIERED',
+}
+
 /**
  * The Amazon MSK configuration to use for the cluster.
  * Note: There is currently no Cloudformation Resource to create a Configuration
@@ -500,6 +522,21 @@ export class Cluster extends ClusterBase {
       inCluster: props.encryptionInTransit?.enableInCluster ?? true,
     };
 
+    const storageMode =
+      props.storageMode ?? StorageMode.LOCAL;
+    if (storageMode === StorageMode.TIERED && KafkaVersion.isTieredStorageCompatible(props.kafkaVersion) === false) {
+      throw Error(
+        'To utilize Tiered storage mode, the MSK cluster Kafka version must be compatiable.',
+      );
+    }
+    if (storageMode === StorageMode.TIERED && instanceType === this.mskInstanceType(
+      ec2.InstanceType.of(ec2.InstanceClass.T3, ec2.InstanceSize.SMALL),
+    )) {
+      throw Error(
+        'The t3.small instance type does not support Tiered storage mode.',
+      );
+    }
+
     const openMonitoring =
       props.monitoring?.enablePrometheusJmxExporter ||
       props.monitoring?.enablePrometheusNodeExporter
@@ -683,6 +720,7 @@ export class Cluster extends ClusterBase {
       configurationInfo: props.configurationInfo,
       enhancedMonitoring: props.monitoring?.clusterMonitoringLevel,
       openMonitoring: openMonitoring,
+      storageMode: storageMode,
       loggingInfo: loggingInfo,
       clientAuthentication: clientAuthentication,
     });

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

@@ -37,11 +37,13 @@ describe('MSK Cluster', () => {
     [msk.KafkaVersion.V2_7_2, '2.7.2'],
     [msk.KafkaVersion.V2_8_0, '2.8.0'],
     [msk.KafkaVersion.V2_8_1, '2.8.1'],
+    [msk.KafkaVersion.V2_8_2_TIERED, '2.8.2.tiered'],
     [msk.KafkaVersion.V3_1_1, '3.1.1'],
     [msk.KafkaVersion.V3_2_0, '3.2.0'],
     [msk.KafkaVersion.V3_3_1, '3.3.1'],
     [msk.KafkaVersion.V3_3_2, '3.3.2'],
     [msk.KafkaVersion.V3_4_0, '3.4.0'],
+    [msk.KafkaVersion.V3_5_1, '3.5.1'],
   ],
   )('created with expected Kafka version %j', (parameter, result) => {
     new msk.Cluster(stack, 'Cluster', {
@@ -242,6 +244,7 @@ describe('MSK Cluster', () => {
               'arn:aws:kms:us-east-1:111122223333:key/1234abc',
             ),
           },
+          storageMode: msk.StorageMode.LOCAL,
           encryptionInTransit: {
             clientBroker: msk.ClientBrokerEncryption.TLS,
           },
@@ -694,6 +697,7 @@ describe('MSK Cluster', () => {
           'arn:aws:kms:us-east-1:111122223333:key/1234abc',
         ),
       },
+      storageMode: msk.StorageMode.LOCAL,
       encryptionInTransit: {
         clientBroker: msk.ClientBrokerEncryption.TLS,
       },
@@ -785,5 +789,62 @@ describe('MSK Cluster', () => {
       });
     });
   });
+
+  describe('created with storage mode', () => {
+    describe('with tiered storage mode', () => {
+      test('fails if incompatiable Kafka version', () => {
+        expect(
+          () =>
+            new msk.Cluster(stack, 'Cluster', {
+              clusterName: 'cluster',
+              kafkaVersion: msk.KafkaVersion.V2_6_1,
+              vpc,
+              storageMode: msk.StorageMode.TIERED,
+            }),
+        ).toThrow(
+          'To utilize Tiered storage mode, the MSK cluster Kafka version must be compatiable.',
+        );
+      });
+      test('fails if instance type of t3.small', () => {
+        expect(
+          () =>
+            new msk.Cluster(stack, 'Cluster', {
+              clusterName: 'cluster',
+              instanceType: ec2.InstanceType.of(ec2.InstanceClass.T3, ec2.InstanceSize.SMALL),
+              kafkaVersion: msk.KafkaVersion.V2_8_2_TIERED,
+              vpc,
+              storageMode: msk.StorageMode.TIERED,
+            }),
+        ).toThrow(
+          'The t3.small instance type does not support Tiered storage mode.',
+        );
+      });
+      test('create a cluster with tiered storage mode', () => {
+        new msk.Cluster(stack, 'Cluster', {
+          clusterName: 'cluster',
+          instanceType: ec2.InstanceType.of(ec2.InstanceClass.M5, ec2.InstanceSize.LARGE),
+          kafkaVersion: msk.KafkaVersion.V2_8_2_TIERED,
+          vpc,
+          storageMode: msk.StorageMode.TIERED,
+        }),
+        Template.fromStack(stack).hasResourceProperties('AWS::MSK::Cluster', {
+          StorageMode: 'TIERED',
+        });
+      });
+    });
+    describe('with local storage mode', () => {
+      test('create a cluster with local storage mode', () => {
+        new msk.Cluster(stack, 'Cluster', {
+          clusterName: 'cluster',
+          kafkaVersion: msk.KafkaVersion.V2_6_1,
+          vpc,
+          storageMode: msk.StorageMode.LOCAL,
+        });
+        Template.fromStack(stack).hasResourceProperties('AWS::MSK::Cluster', {
+          StorageMode: 'LOCAL',
+        });
+      });
+    });
+  });
 });
 

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

@@ -12,7 +12,7 @@ const vpc = new Vpc(stack, 'Vpc', { maxAzs: 2, restrictDefaultSecurityGroup: fal
 const cluster = new Cluster(stack, 'Cluster', {
   vpc,
   clusterName: 'integ-test',
-  kafkaVersion: KafkaVersion.V3_4_0,
+  kafkaVersion: KafkaVersion.V3_5_1,
   removalPolicy: RemovalPolicy.DESTROY,
   clientAuthentication: ClientAuthentication.sasl({ scram: true }),
 });

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

@@ -19,11 +19,13 @@ const versions: KafkaVersion[] = [
   KafkaVersion.V2_7_2,
   KafkaVersion.V2_8_0,
   KafkaVersion.V2_8_1,
+  KafkaVersion.V2_8_2_TIERED,
   KafkaVersion.V3_1_1,
   KafkaVersion.V3_2_0,
   KafkaVersion.V3_3_1,
   KafkaVersion.V3_3_2,
   KafkaVersion.V3_4_0,
+  KafkaVersion.V3_5_1,
 ];
 
 class KafkaVersionTest extends Stack {

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

@@ -19,7 +19,7 @@ class KafkaZookeeperTest extends Stack {
 
     const cluster = new msk.Cluster(this, 'ClusterZookeeper', {
       clusterName: 'cluster-zookeeper',
-      kafkaVersion: msk.KafkaVersion.V3_4_0,
+      kafkaVersion: msk.KafkaVersion.V3_5_1,
       vpc,
       removalPolicy: RemovalPolicy.DESTROY,
     });