feat(kinesis): support stream consumers

- introduce `StreamConsumer` construct to model `AWS::Kinesis::StreamConsumer`
  - introduce `addToResourcePolicy` to enable creating/configuring a resource policy for the consumer
  - introduce `grant` and `grantRead` for granting permissions
  - leverage `iam.Grant.addToPrincipalOrResource` in `grant` to be able to use `grant` methods cross environments to update the grantee's iam policy and the consumer's resource policy as needed
- update `ResourcePolicy` to support both `Stream` and `StreamConsumer`
- update `Stream`'s `grant` to leverage `iam.Grant.addToPrincipalOrResource` for cross-environment support

closes #32050

packages/@aws-cdk-testing/framework-integ/test/aws-kinesis/test/integ.resource-policy.js.snapshot/kinesis-resource-policy.template.json

@@ -67,6 +67,65 @@
      "Version": "2012-10-17"
     }
    }
+  },
+  "StreamConsumer58240CBA": {
+   "Type": "AWS::Kinesis::StreamConsumer",
+   "Properties": {
+    "ConsumerName": "stream-consumer",
+    "StreamARN": {
+     "Fn::GetAtt": [
+      "MyStream5C050E93",
+      "Arn"
+     ]
+    }
+   }
+  },
+  "StreamConsumerPolicy925BAE36": {
+   "Type": "AWS::Kinesis::ResourcePolicy",
+   "Properties": {
+    "ResourceArn": {
+     "Fn::GetAtt": [
+      "StreamConsumer58240CBA",
+      "ConsumerARN"
+     ]
+    },
+    "ResourcePolicy": {
+     "Statement": [
+      {
+       "Action": [
+        "kinesis:DescribeStreamConsumer",
+        "kinesis:SubscribeToShard"
+       ],
+       "Effect": "Allow",
+       "Principal": {
+        "AWS": {
+         "Fn::Join": [
+          "",
+          [
+           "arn:",
+           {
+            "Ref": "AWS::Partition"
+           },
+           ":iam::",
+           {
+            "Ref": "AWS::AccountId"
+           },
+           ":root"
+          ]
+         ]
+        }
+       },
+       "Resource": {
+        "Fn::GetAtt": [
+         "StreamConsumer58240CBA",
+         "ConsumerARN"
+        ]
+       }
+      }
+     ],
+     "Version": "2012-10-17"
+    }
+   }
   }
  },
  "Conditions": {

packages/@aws-cdk-testing/framework-integ/test/aws-kinesis/test/integ.resource-policy.ts

@@ -1,5 +1,5 @@
 import { App, Stack } from 'aws-cdk-lib';
-import { Stream } from 'aws-cdk-lib/aws-kinesis';
+import { Stream, StreamConsumer } from 'aws-cdk-lib/aws-kinesis';
 import { AccountPrincipal, PolicyStatement } from 'aws-cdk-lib/aws-iam';
 import { IntegTest } from '@aws-cdk/integ-tests-alpha';
 
@@ -8,6 +8,11 @@ const stack = new Stack(app, 'kinesis-resource-policy');
 
 const stream = new Stream(stack, 'MyStream');
 
+const streamConsumer = new StreamConsumer(stack, 'StreamConsumer', {
+  streamConsumerName: 'stream-consumer',
+  stream: stream,
+});
+
 stream.addToResourcePolicy(new PolicyStatement({
   resources: [stream.streamArn],
   actions: [
@@ -17,6 +22,15 @@ stream.addToResourcePolicy(new PolicyStatement({
   principals: [new AccountPrincipal(stack.account)],
 }));
 
+streamConsumer.addToResourcePolicy(new PolicyStatement({
+  resources: [streamConsumer.streamConsumerArn],
+  actions: [
+    'kinesis:DescribeStreamConsumer',
+    'kinesis:SubscribeToShard',
+  ],
+  principals: [new AccountPrincipal(stack.account)],
+}));
+
 new IntegTest(app, 'integ-kinesis-resource-policy', {
   testCases: [stack],
   stackUpdateWorkflow: false,