feat(scheduler-targets-alpha): KinesisStreamPutRecord Target (#27845)

This PR adds KinesisStreamPutRecord Target for EventBridge Scheduler.

Closes #27451.

----

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

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

@@ -31,6 +31,7 @@ The following targets are supported:
 5. `targets.SnsPublish`: [Publish messages to an Amazon SNS topic](#publish-messages-to-an-amazon-sns-topic)
 6. `targets.EventBridgePutEvents`: [Put Events on EventBridge](#send-events-to-an-eventbridge-event-bus)
 7. `targets.InspectorStartAssessmentRun`: [Start an Amazon Inspector assessment run](#start-an-amazon-inspector-assessment-run)
+8. `targets.KinesisStreamPutRecord`: [Put a record to an Amazon Kinesis Data Streams](#put-a-record-to-an-amazon-kinesis-data-streams)
 
 ## Invoke a Lambda function
 
@@ -225,3 +226,23 @@ new Schedule(this, 'Schedule', {
   target: new targets.InspectorStartAssessmentRun(assessmentTemplate),
 });
 ```
+
+## Put a record to an Amazon Kinesis Data Streams
+
+Use the `KinesisStreamPutRecord` target to put a record to an Amazon Kinesis Data Streams.
+
+The code snippet below creates an event rule with a stream as target which is
+called every hour by Event Bridge Scheduler.
+
+```ts
+import * as kinesis from 'aws-cdk-lib/aws-kinesis';
+
+const stream = new kinesis.Stream(this, 'MyStream');
+
+new Schedule(this, 'Schedule', {
+  schedule: ScheduleExpression.rate(Duration.minutes(60)),
+  target: new targets.KinesisStreamPutRecord(stream, {
+    partitionKey: 'key',
+  }),
+});
+```

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

@@ -1,6 +1,7 @@
 export * from './codebuild-start-build';
 export * from './event-bridge-put-events';
 export * from './inspector-start-assessment-run';
+export * from './kinesis-stream-put-record';
 export * from './lambda-invoke';
 export * from './sns-publish';
 export * from './sqs-send-message';

packages/@aws-cdk/aws-scheduler-targets-alpha/lib/kinesis-stream-put-record.ts

@@ -0,0 +1,61 @@
+import { ISchedule, IScheduleTarget, ScheduleTargetConfig } from '@aws-cdk/aws-scheduler-alpha';
+import { Names, Token } from 'aws-cdk-lib';
+import { IRole } from 'aws-cdk-lib/aws-iam';
+import * as kinesis from 'aws-cdk-lib/aws-kinesis';
+import { ScheduleTargetBase, ScheduleTargetBaseProps } from './target';
+import { sameEnvDimension } from './util';
+
+/**
+ * Properties for a Kinesis Data Streams Target
+ */
+export interface KinesisStreamPutRecordProps extends ScheduleTargetBaseProps {
+  /**
+   * The shard to which EventBridge Scheduler sends the event.
+   *
+   * The length must be between 1 and 256.
+   *
+   * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-scheduler-schedule-kinesisparameters.html
+   */
+  readonly partitionKey: string;
+}
+
+/**
+ * Use an Amazon Kinesis Data Streams as a target for AWS EventBridge Scheduler.
+ */
+export class KinesisStreamPutRecord extends ScheduleTargetBase implements IScheduleTarget {
+  constructor(
+    private readonly stream: kinesis.IStream,
+    private readonly props: KinesisStreamPutRecordProps,
+  ) {
+    super(props, stream.streamArn);
+
+    if (!Token.isUnresolved(props.partitionKey) && (props.partitionKey.length < 1 || props.partitionKey.length > 256)) {
+      throw new Error(`partitionKey length must be between 1 and 256, got ${props.partitionKey.length}`);
+    }
+  }
+
+  protected addTargetActionToRole(schedule: ISchedule, role: IRole): void {
+    if (!sameEnvDimension(this.stream.env.region, schedule.env.region)) {
+      throw new Error(`Cannot assign stream in region ${this.stream.env.region} to the schedule ${Names.nodeUniqueId(schedule.node)} in region ${schedule.env.region}. Both the schedule and the stream must be in the same region.`);
+    }
+
+    if (!sameEnvDimension(this.stream.env.account, schedule.env.account)) {
+      throw new Error(`Cannot assign stream in account ${this.stream.env.account} to the schedule ${Names.nodeUniqueId(schedule.node)} in account ${schedule.env.region}. Both the schedule and the stream must be in the same account.`);
+    }
+
+    if (this.props.role && !sameEnvDimension(this.props.role.env.account, this.stream.env.account)) {
+      throw new Error(`Cannot grant permission to execution role in account ${this.props.role.env.account} to invoke target ${Names.nodeUniqueId(this.stream.node)} in account ${this.stream.env.account}. Both the target and the execution role must be in the same account.`);
+    }
+
+    this.stream.grantWrite(role);
+  }
+
+  protected bindBaseTargetConfig(_schedule: ISchedule): ScheduleTargetConfig {
+    return {
+      ...super.bindBaseTargetConfig(_schedule),
+      kinesisParameters: {
+        partitionKey: this.props.partitionKey,
+      },
+    };
+  }
+}