feat(aws-kinesis): add support for data streams capacity modes (#18074)

The latest CloudFormation spec (v51) added support for selecting a capacity mode for Kinesis Data Streams. This is already supported by the generated L1 construct. This change adds support for the feature to the L2 construct class `Stream` via the constructor props `StreamProps`.

closes #18050

----

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

Author: igilham

packages/@aws-cdk/aws-kinesis/lib/stream.ts

@@ -699,6 +699,13 @@ export interface StreamProps {
    *   will be created and associated with this stream.
    */
   readonly encryptionKey?: kms.IKey;
+
+  /**
+   * The capacity mode of this stream.
+   *
+   * @default StreamMode.PROVISIONED
+   */
+  readonly streamMode?: StreamMode;
 }
 
 /**
@@ -746,6 +753,9 @@ export class Stream extends StreamBase {
     });
 
     const shardCount = props.shardCount || 1;
+
+    const streamMode = props.streamMode;
+
     const retentionPeriodHours = props.retentionPeriod?.toHours() ?? 24;
     if (!Token.isUnresolved(retentionPeriodHours)) {
       if (retentionPeriodHours < 24 || retentionPeriodHours > 8760) {
@@ -760,6 +770,7 @@ export class Stream extends StreamBase {
       retentionPeriodHours,
       shardCount,
       streamEncryption,
+      streamModeDetails: streamMode ? { streamMode } : undefined,
     });
 
     this.streamArn = this.getResourceArnAttribute(this.stream.attrArn, {
@@ -858,3 +869,20 @@ export enum StreamEncryption {
    */
   MANAGED = 'MANAGED'
 }
+
+/**
+ * Specifies the capacity mode to apply to this stream.
+ */
+export enum StreamMode {
+  /**
+   * Specify the provisioned capacity mode. The stream will have `shardCount` shards unless
+   * modified and will be billed according to the provisioned capacity.
+   */
+  PROVISIONED = 'PROVISIONED',
+
+  /**
+   * Specify the on-demand capacity mode. The stream will autoscale and be billed according to the
+   * volume of data ingested and retrieved.
+   */
+  ON_DEMAND = 'ON_DEMAND'
+}

packages/@aws-cdk/aws-kinesis/test/stream.test.ts

@@ -5,7 +5,7 @@ import * as kms from '@aws-cdk/aws-kms';
 import { testFutureBehavior, testLegacyBehavior } from '@aws-cdk/cdk-build-tools';
 import { App, Duration, Stack, CfnParameter } from '@aws-cdk/core';
 import * as cxapi from '@aws-cdk/cx-api';
-import { Stream, StreamEncryption } from '../lib';
+import { Stream, StreamEncryption, StreamMode } from '../lib';
 
 /* eslint-disable quote-props */
 
@@ -374,6 +374,63 @@ describe('Kinesis data streams', () => {
     });
   }),
 
+  test.each([StreamMode.ON_DEMAND, StreamMode.PROVISIONED])('uses explicit capacity mode %s', (mode: StreamMode) => {
+    const stack = new Stack();
+
+    new Stream(stack, 'MyStream', {
+      streamMode: mode,
+    });
+
+    expect(stack).toMatchTemplate({
+      Resources: {
+        MyStream5C050E93: {
+          Type: 'AWS::Kinesis::Stream',
+          Properties: {
+            ShardCount: 1,
+            RetentionPeriodHours: 24,
+            StreamModeDetails: {
+              StreamMode: StreamMode[mode],
+            },
+            StreamEncryption: {
+              'Fn::If': [
+                'AwsCdkKinesisEncryptedStreamsUnsupportedRegions',
+                {
+                  Ref: 'AWS::NoValue',
+                },
+                {
+                  EncryptionType: 'KMS',
+                  KeyId: 'alias/aws/kinesis',
+                },
+              ],
+            },
+          },
+        },
+      },
+      Conditions: {
+        AwsCdkKinesisEncryptedStreamsUnsupportedRegions: {
+          'Fn::Or': [
+            {
+              'Fn::Equals': [
+                {
+                  Ref: 'AWS::Region',
+                },
+                'cn-north-1',
+              ],
+            },
+            {
+              'Fn::Equals': [
+                {
+                  Ref: 'AWS::Region',
+                },
+                'cn-northwest-1',
+              ],
+            },
+          ],
+        },
+      },
+    });
+  });
+
   test('grantRead creates and attaches a policy with read only access to the principal', () => {
     const stack = new Stack();
     const stream = new Stream(stack, 'MyStream', {