feat(pipes-targets): add Kinesis (#30656)

Add Kinesis data stream as a Pipes target.

It's nontrivial to get the data from the Kinesis data stream, but here are screenshots 
showing data made it through during the integration test.

<img width="656" alt="Screenshot 2024-06-24 at 7 24 10 PM" src="https://github.com/aws/aws-cdk/assets/3310356/bc6e12a2-8fea-42a7-baaa-e8b5b5ea652f">

<img width="649" alt="Screenshot 2024-06-24 at 7 26 35 PM" src="https://github.com/aws/aws-cdk/assets/3310356/5224b0d9-a356-47e6-ab48-3551ff3b5078">

packages/@aws-cdk/aws-pipes-targets-alpha/README.md

@@ -33,6 +33,7 @@ The following targets are supported:
 2. `targets.SfnStateMachine`: [Invoke a State Machine from an event source](#aws-step-functions-state-machine)
 3. `targets.LambdaFunction`: [Send event source to a Lambda Function](#aws-lambda-function)
 4. `targets.ApiDestinationTarget`: [Send event source to an EventBridge API Destination](#amazon-eventbridge-api-destination)
+5. `targets.KinesisTarget`: [Send event source to a Kinesis data stream](#amazon-kinesis-data-stream)
 
 ### Amazon SQS
 
@@ -205,3 +206,38 @@ const pipe = new pipes.Pipe(this, 'Pipe', {
     target: apiTarget,
 });
 ```
+
+### Amazon Kinesis Data Stream
+
+A data stream can be used as a target for a pipe. The data stream will receive the (enriched/filtered) source payload.
+
+```ts
+declare const sourceQueue: sqs.Queue;
+declare const targetStream: kinesis.Stream;
+
+const streamTarget = new targets.KinesisTarget(targetStream, {
+    partitionKey: 'pk',
+});
+
+const pipe = new pipes.Pipe(this, 'Pipe', {
+    source: new SqsSource(sourceQueue),
+    target: streamTarget,
+});
+```
+
+The input to the target data stream can be transformed:
+
+```ts
+declare const sourceQueue: sqs.Queue;
+declare const targetStream: kinesis.Stream;
+
+const streamTarget = new targets.KinesisTarget(targetStream, {
+    partitionKey: 'pk',
+    inputTransformation: pipes.InputTransformation.fromObject({ body: "👀" }),
+});
+
+const pipe = new pipes.Pipe(this, 'Pipe', {
+    source: new SqsSource(sourceQueue),
+    target: streamTarget,
+});
+```

packages/@aws-cdk/aws-pipes-targets-alpha/lib/index.ts

@@ -1,4 +1,5 @@
 export * from './api-destination';
+export * from './kinesis';
 export * from './lambda';
 export * from './sqs';
 export * from './stepfunctions';

packages/@aws-cdk/aws-pipes-targets-alpha/lib/kinesis.ts

@@ -0,0 +1,60 @@
+import { IInputTransformation, IPipe, ITarget, TargetConfig } from '@aws-cdk/aws-pipes-alpha';
+import { Token } from 'aws-cdk-lib';
+import { IRole } from 'aws-cdk-lib/aws-iam';
+import { IStream } from 'aws-cdk-lib/aws-kinesis';
+
+/**
+ * Kinesis target properties.
+ */
+export interface KinesisTargetParameters {
+  /**
+   * The input transformation to apply to the message before sending it to the target.
+   *
+   * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipetargetparameters.html#cfn-pipes-pipe-pipetargetparameters-inputtemplate
+   * @default - none
+   */
+  readonly inputTransformation?: IInputTransformation;
+
+  /**
+   * Determines which shard in the stream the data record is assigned to.
+   *
+   * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipetargetkinesisstreamparameters.html#cfn-pipes-pipe-pipetargetkinesisstreamparameters-partitionkey
+   */
+  readonly partitionKey: string;
+}
+
+/**
+ * An EventBridge Pipes target that sends messages to a Kinesis stream.
+ */
+export class KinesisTarget implements ITarget {
+  private stream: IStream;
+  private streamParameters: KinesisTargetParameters;
+  public readonly targetArn: string;
+
+  constructor(stream: IStream, parameters: KinesisTargetParameters) {
+    this.stream = stream;
+    this.targetArn = stream.streamArn;
+    this.streamParameters = parameters;
+
+    validatePartitionKey(parameters.partitionKey);
+  }
+
+  grantPush(grantee: IRole): void {
+    this.stream.grantWrite(grantee);
+  }
+
+  bind(pipe: IPipe): TargetConfig {
+    return {
+      targetParameters: {
+        inputTemplate: this.streamParameters.inputTransformation?.bind(pipe).inputTemplate,
+        kinesisStreamParameters: this.streamParameters,
+      },
+    };
+  }
+}
+
+function validatePartitionKey(pk: string) {
+  if (!Token.isUnresolved(pk) && pk.length > 256) {
+    throw new Error(`Partition key must be less than or equal to 256 characters, received ${pk.length}`);
+  }
+}

packages/@aws-cdk/aws-pipes-targets-alpha/rosetta/default.ts-fixture

@@ -1,6 +1,7 @@
 // Fixture with packages imported, but nothing else
 import * as cdk from 'aws-cdk-lib';
 import * as events from 'aws-cdk-lib/aws-events';
+import * as kinesis from 'aws-cdk-lib/aws-kinesis';
 import * as sqs from 'aws-cdk-lib/aws-sqs';
 import * as sfn from 'aws-cdk-lib/aws-stepfunctions';
 import * as lambda from 'aws-cdk-lib/aws-lambda';

packages/@aws-cdk/aws-pipes-targets-alpha/test/__snapshots__/kinesis.test.ts.snap

@@ -0,0 +1,58 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`Kinesis should grant pipe role push access 1`] = `
+{
+  "MyPipeRoleCBC8E9AB": {
+    "Properties": {
+      "AssumeRolePolicyDocument": {
+        "Statement": [
+          {
+            "Action": "sts:AssumeRole",
+            "Effect": "Allow",
+            "Principal": {
+              "Service": "pipes.amazonaws.com",
+            },
+          },
+        ],
+        "Version": "2012-10-17",
+      },
+    },
+    "Type": "AWS::IAM::Role",
+  },
+}
+`;
+
+exports[`Kinesis should grant pipe role push access 2`] = `
+{
+  "MyPipeRoleDefaultPolicy31387C20": {
+    "Properties": {
+      "PolicyDocument": {
+        "Statement": [
+          {
+            "Action": [
+              "kinesis:ListShards",
+              "kinesis:PutRecord",
+              "kinesis:PutRecords",
+            ],
+            "Effect": "Allow",
+            "Resource": {
+              "Fn::GetAtt": [
+                "MyStream5C050E93",
+                "Arn",
+              ],
+            },
+          },
+        ],
+        "Version": "2012-10-17",
+      },
+      "PolicyName": "MyPipeRoleDefaultPolicy31387C20",
+      "Roles": [
+        {
+          "Ref": "MyPipeRoleCBC8E9AB",
+        },
+      ],
+    },
+    "Type": "AWS::IAM::Policy",
+  },
+}
+`;