# Amazon Kinesis Construct Library

[Amazon Kinesis](https://docs.aws.amazon.com/streams/latest/dev/introduction.html) provides collection and processing of large
[streams](https://aws.amazon.com/streaming-data/) of data records in real time. Kinesis data streams can be used for rapid and continuous data
intake and aggregation.

## Streams

Amazon Kinesis Data Streams ingests a large amount of data in real time, durably stores the data, and makes the data available for consumption.

Using the CDK, a new Kinesis stream can be created as part of the stack using the construct's constructor. You may specify the `streamName` to give
your own identifier to the stream. If not, CloudFormation will generate a name.

```ts
new kinesis.Stream(this, 'MyFirstStream', {
  streamName: 'my-awesome-stream',
});
```

You can also specify properties such as `shardCount` to indicate how many shards the stream should choose and a `retentionPeriod`
to specify how long the data in the shards should remain accessible.
Read more at [Creating and Managing Streams](https://docs.aws.amazon.com/streams/latest/dev/working-with-streams.html)

```ts
new kinesis.Stream(this, 'MyFirstStream', {
  streamName: 'my-awesome-stream',
  shardCount: 3,
  retentionPeriod: Duration.hours(48),
});
```

### Encryption

[Stream encryption](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-kinesis-stream-streamencryption.html) enables
server-side encryption using an AWS KMS key for a specified stream.

Encryption is enabled by default on your stream with the master key owned by Kinesis Data Streams in regions where it is supported.

```ts
new kinesis.Stream(this, 'MyEncryptedStream');
```

You can enable encryption on your stream with a user-managed key by specifying the `encryption` property.
A KMS key will be created for you and associated with the stream.

```ts
new kinesis.Stream(this, 'MyEncryptedStream', {
  encryption: kinesis.StreamEncryption.KMS,
});
```

You can also supply your own external KMS key to use for stream encryption by specifying the `encryptionKey` property.

```ts
const key = new kms.Key(this, 'MyKey');

new kinesis.Stream(this, 'MyEncryptedStream', {
  encryption: kinesis.StreamEncryption.KMS,
  encryptionKey: key,
});
```

### Import

Any Kinesis stream that has been created outside the stack can be imported into your CDK app.

Streams can be imported by their ARN via the `Stream.fromStreamArn()` API

```ts
const importedStream = kinesis.Stream.fromStreamArn(this, 'ImportedStream',
  'arn:aws:kinesis:us-east-2:123456789012:stream/f3j09j2230j',
);
```

Encrypted Streams can also be imported by their attributes via the `Stream.fromStreamAttributes()` API

```ts
const importedStream = kinesis.Stream.fromStreamAttributes(this, 'ImportedEncryptedStream', {
  streamArn: 'arn:aws:kinesis:us-east-2:123456789012:stream/f3j09j2230j',
  encryptionKey: kms.Key.fromKeyArn(this, 'key',
    'arn:aws:kms:us-east-1:123456789012:key/12345678-1234-1234-1234-123456789012',
  ),
});
```

### Permission Grants

IAM roles, users or groups which need to be able to work with Amazon Kinesis streams at runtime should be granted IAM permissions.

Any object that implements the `IGrantable` interface (has an associated principal) can be granted permissions by calling:

- `grantRead(principal)` - grants the principal read access
- `grantWrite(principal)` - grants the principal write permissions to a Stream
- `grantReadWrite(principal)` - grants principal read and write permissions

#### Read Permissions

Grant `read` access to a stream by calling the `grantRead()` API.
If the stream has an encryption key, read permissions will also be granted to the key.

```ts
const lambdaRole = new iam.Role(this, 'Role', {
  assumedBy: new iam.ServicePrincipal('lambda.amazonaws.com'),
  description: 'Example role...',
});

const stream = new kinesis.Stream(this, 'MyEncryptedStream', {
  encryption: kinesis.StreamEncryption.KMS,
});

// give lambda permissions to read stream
stream.grantRead(lambdaRole);
```

The following read permissions are provided to a service principal by the `grantRead()` API:

- `kinesis:DescribeStreamSummary`
- `kinesis:GetRecords`
- `kinesis:GetShardIterator`
- `kinesis:ListShards`
- `kinesis:SubscribeToShard`

#### Write Permissions

Grant `write` permissions to a stream is provided by calling the `grantWrite()` API.
If the stream has an encryption key, write permissions will also be granted to the key.

```ts
const lambdaRole = new iam.Role(this, 'Role', {
  assumedBy: new iam.ServicePrincipal('lambda.amazonaws.com'),
  description: 'Example role...',
});

const stream = new kinesis.Stream(this, 'MyEncryptedStream', {
  encryption: kinesis.StreamEncryption.KMS,
});

// give lambda permissions to write to stream
stream.grantWrite(lambdaRole);
```

The following write permissions are provided to a service principal by the `grantWrite()` API:

- `kinesis:ListShards`
- `kinesis:PutRecord`
- `kinesis:PutRecords`

#### Custom Permissions

You can add any set of permissions to a stream by calling the `grant()` API.

```ts
const user = new iam.User(this, 'MyUser');

const stream = new kinesis.Stream(this, 'MyStream');

// give my user permissions to list shards
stream.grant(user, 'kinesis:ListShards');
```

### Metrics

You can use common metrics from your stream to create alarms and/or dashboards. The `stream.metric('MetricName')` method creates a metric with the stream namespace and dimension. You can also use pre-define methods like `stream.metricGetRecordsSuccess()`. To find out more about Kinesis metrics check [Monitoring the Amazon Kinesis Data Streams Service with Amazon CloudWatch](https://docs.aws.amazon.com/streams/latest/dev/monitoring-with-cloudwatch.html).

```ts
const stream = new kinesis.Stream(this, 'MyStream');

// Using base metric method passing the metric name
stream.metric('GetRecords.Success');

// using pre-defined metric method
stream.metricGetRecordsSuccess();

// using pre-defined and overriding the statistic
stream.metricGetRecordsSuccess({ statistic: 'Maximum' });
```

#### Shard-level Metrics

You can enable enhanced shard-level metrics for your Kinesis stream to get detailed monitoring of individual shards. Shard-level metrics provide more granular insights into the performance and health of your stream.

```ts
const stream = new kinesis.Stream(this, 'MyStream', {
  shardLevelMetrics: [kinesis.ShardLevelMetrics.ALL],
});
```

You can also specify individual metrics that you want to monitor:

```ts
const stream = new kinesis.Stream(this, 'MyStream', {
  shardLevelMetrics: [
    kinesis.ShardLevelMetrics.INCOMING_BYTES,
    kinesis.ShardLevelMetrics.INCOMING_RECORDS,
    kinesis.ShardLevelMetrics.ITERATOR_AGE_MILLISECONDS,
    kinesis.ShardLevelMetrics.OUTGOING_BYTES,
    kinesis.ShardLevelMetrics.OUTGOING_RECORDS,
    kinesis.ShardLevelMetrics.READ_PROVISIONED_THROUGHPUT_EXCEEDED,
    kinesis.ShardLevelMetrics.WRITE_PROVISIONED_THROUGHPUT_EXCEEDED,
  ],
});
```

Available shard-level metrics include:

- `INCOMING_BYTES` - The number of bytes successfully put to the shard
- `INCOMING_RECORDS` - The number of records successfully put to the shard
- `ITERATOR_AGE_MILLISECONDS` - The age of the last record in all GetRecords calls made against a shard
- `OUTGOING_BYTES` - The number of bytes retrieved from the shard
- `OUTGOING_RECORDS` - The number of records retrieved from the shard
- `READ_PROVISIONED_THROUGHPUT_EXCEEDED` - The number of GetRecords calls throttled for the shard
- `WRITE_PROVISIONED_THROUGHPUT_EXCEEDED` - The number of records rejected due to throttling for the shard
- `ALL` - All available metrics

Note: You cannot specify `ALL` together with other individual metrics. If you want all metrics, use `ALL` alone.

For more information about shard-level metrics, see [Monitoring the Amazon Kinesis Data Streams Service with Amazon CloudWatch](https://docs.aws.amazon.com/streams/latest/dev/monitoring-with-cloudwatch.html#kinesis-metrics-shard).

## Stream Consumers

Creating stream consumers allow consumers to receive data from the stream using enhanced fan-out at a rate of up to 2 MiB per second for every shard.
This rate is unaffected by the total number of consumers that read from the same stream.

For more information, see [Develop enhanced fan-out consumers with dedicated throughput](https://docs.aws.amazon.com/streams/latest/dev/enhanced-consumers.html).

To create and associate a stream consumer with a stream

```ts
const stream = new kinesis.Stream(this, 'MyStream');

const streamConsumer = new kinesis.StreamConsumer(this, 'MyStreamConsumer', {
  streamConsumerName: 'MyStreamConsumer',
  stream,
});
```

#### Read Permissions

Grant `read` access to a stream consumer, and the stream it is registered with, by calling the `grantRead()` API.

```ts
const lambdaRole = new iam.Role(this, 'Role', {
  assumedBy: new iam.ServicePrincipal('lambda.amazonaws.com'),
  description: 'Example role...',
});

const stream = new kinesis.Stream(this, 'MyEncryptedStream', {
  encryption: kinesis.StreamEncryption.KMS,
});
const streamConsumer = new kinesis.StreamConsumer(this, 'MyStreamConsumer', {
  streamConsumerName: 'MyStreamConsumer',
  stream,
});

// give lambda permissions to read stream via the stream consumer
streamConsumer.grantRead(lambdaRole);
```

In addition to stream's permissions, the following permissions are provided to a service principal by the `grantRead()` API:

- `kinesis:DescribeStreamConsumer`,
- `kinesis:SubscribeToShard`,

## Resource Policy

You can create a resource policy for a data stream or a stream consumer.
For more information, see [Controlling access to Amazon Kinesis Data Streams resources using IAM](https://docs.aws.amazon.com/streams/latest/dev/controlling-access.html).

A resource policy is automatically created when `addToResourcePolicy` is called, if one doesn't already exist.

Using `addToResourcePolicy` is the simplest way to add a resource policy:

```ts
const stream = new kinesis.Stream(this, 'MyStream');
const streamConsumer = new kinesis.StreamConsumer(this, 'MyStreamConsumer', {
  streamConsumerName: 'MyStreamConsumer',
  stream,
});

// create a stream resource policy via addToResourcePolicy method
stream.addToResourcePolicy(new iam.PolicyStatement({
  resources: [stream.streamArn],
  actions: ['kinesis:GetRecords'],
  principals: [new iam.AnyPrincipal()],
}));

// create a stream consumer resource policy via addToResourcePolicy method
streamConsumer.addToResourcePolicy(new iam.PolicyStatement({
  resources: [stream.streamArn],
  actions: ['kinesis:DescribeStreamConsumer'],
  principals: [new iam.AnyPrincipal()],
}));
```

You can create a resource manually by using [`ResourcePolicy`](/v2/docs/aws-cdk-lib.aws_kinesis.ResourcePolicy).
Also, you can set a custom policy document to [`ResourcePolicy`](/v2/docs/aws-cdk-lib.aws_kinesis.ResourcePolicy).
If not, a blank policy document will be set.

```ts
const stream = new kinesis.Stream(this, 'MyStream');
const streamConsumer = new kinesis.StreamConsumer(this, 'MyStreamConsumer', {
  streamConsumerName: 'MyStreamConsumer',
  stream,
});

// create a custom policy document
const policyDocument = new iam.PolicyDocument({
  assignSids: true,
  statements: [
    new iam.PolicyStatement({
      actions: ['kinesis:GetRecords'],
      resources: [stream.streamArn],
      principals: [new iam.AnyPrincipal()],
    }),
  ],
});

// create a stream resource policy manually
new kinesis.ResourcePolicy(this, 'ResourcePolicy', {
  stream,
  policyDocument,
});

// create a stream consumer resource policy manually
new kinesis.ResourcePolicy(this, 'ResourcePolicy', {
  streamConsumer,
  policyDocument,
});
```

## Types

| Name | Kind | Summary |
| --- | --- | --- |
| [`ResourcePolicy`](/v2/docs/aws-cdk-lib.aws_kinesis.ResourcePolicy) | Construct (L2) | The policy for a data stream or registered consumer. |
| [`Stream`](/v2/docs/aws-cdk-lib.aws_kinesis.Stream) | Construct (L2) | A Kinesis stream. |
| [`StreamConsumer`](/v2/docs/aws-cdk-lib.aws_kinesis.StreamConsumer) | Construct (L2) | A Kinesis Stream Consumer. |
| [`CfnChannel`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnChannel) | Construct (L1) | Resource Type definition for AWS::Kinesis::Channel. |
| [`CfnResourcePolicy`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnResourcePolicy) | Construct (L1) | Attaches a resource-based policy to a data stream or registered consumer. |
| [`CfnStream`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnStream) | Construct (L1) | Creates a Kinesis stream that captures and transports data records that are emitted from data sources. |
| [`CfnStreamConsumer`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnStreamConsumer) | Construct (L1) | Use the AWS CloudFormation `AWS::Kinesis::StreamConsumer` resource to register a consumer with a Kinesis data stream. |
| [`CfnChannel.CloudWatchLogsConfigurationProperty`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnChannel.CloudWatchLogsConfigurationProperty) | Struct | CloudWatch Logs configuration for the channel. |
| [`CfnChannel.DeadLetterQueueS3ConfigurationProperty`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnChannel.DeadLetterQueueS3ConfigurationProperty) | Struct | Configuration of the S3 bucket used to capture records that cannot be delivered to the primary destination. |
| [`CfnChannel.EncryptionConfigurationProperty`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnChannel.EncryptionConfigurationProperty) | Struct | Server-side encryption configuration for data at rest in the destination. |
| [`CfnChannel.LoggingConfigurationProperty`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnChannel.LoggingConfigurationProperty) | Struct | Configuration for delivering channel operational logs. |
| [`CfnChannel.PartitionFieldProperty`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnChannel.PartitionFieldProperty) | Struct | A single partition field consisting of a transform applied to a source column. |
| [`CfnChannel.PartitionSpecProperty`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnChannel.PartitionSpecProperty) | Struct | Partitioning specification for the destination Iceberg table. |
| [`CfnChannel.RecordConfigurationProperty`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnChannel.RecordConfigurationProperty) | Struct | The configuration that describes how records on the source stream are encoded. |
| [`CfnChannel.S3DestinationConfigurationProperty`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnChannel.S3DestinationConfigurationProperty) | Struct | Configuration for delivery to a vanilla S3 bucket destination. |
| [`CfnChannel.S3StorageConfigurationProperty`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnChannel.S3StorageConfigurationProperty) | Struct | S3 storage configuration that describes the destination bucket and how delivered objects are stored. |
| [`CfnChannel.S3TableConfigurationProperty`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnChannel.S3TableConfigurationProperty) | Struct | An S3 Tables destination including its table bucket, namespace, table name, and partition spec. |
| [`CfnChannel.S3TablesDestinationConfigurationProperty`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnChannel.S3TablesDestinationConfigurationProperty) | Struct | Configuration for delivery to S3 Tables destinations. |
| [`CfnChannel.StreamConfigurationProperty`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnChannel.StreamConfigurationProperty) | Struct | The configuration of a Kinesis stream that the channel reads from. |
| [`CfnChannelProps`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnChannelProps) | Struct | Properties for defining a `CfnChannel`. |
| [`CfnResourcePolicyProps`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnResourcePolicyProps) | Struct | Properties for defining a `CfnResourcePolicy`. |
| [`CfnStream.StreamEncryptionProperty`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnStream.StreamEncryptionProperty) | Struct | Enables or updates server-side encryption using an AWS KMS key for a specified stream. |
| [`CfnStream.StreamModeDetailsProperty`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnStream.StreamModeDetailsProperty) | Struct | Specifies the capacity mode to which you want to set your data stream. |
| [`CfnStream.WarmThroughputObjectProperty`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnStream.WarmThroughputObjectProperty) | Struct | Represents the warm throughput configuration on the stream. |
| [`CfnStreamConsumerProps`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnStreamConsumerProps) | Struct | Properties for defining a `CfnStreamConsumer`. |
| [`CfnStreamProps`](/v2/docs/aws-cdk-lib.aws_kinesis.CfnStreamProps) | Struct | Properties for defining a `CfnStream`. |
| [`IStream`](/v2/docs/aws-cdk-lib.aws_kinesis.IStream) | Interface | A Kinesis Stream. |
| [`IStreamConsumer`](/v2/docs/aws-cdk-lib.aws_kinesis.IStreamConsumer) | Interface | A Kinesis Stream Consumer. |
| [`ResourcePolicyProps`](/v2/docs/aws-cdk-lib.aws_kinesis.ResourcePolicyProps) | Struct | Properties to associate a data stream with a policy. |
| [`ShardLevelMetrics`](/v2/docs/aws-cdk-lib.aws_kinesis.ShardLevelMetrics) | Enum | Enhanced shard-level metrics. |
| [`StreamAttributes`](/v2/docs/aws-cdk-lib.aws_kinesis.StreamAttributes) | Struct | A reference to a stream. |
| [`StreamConsumerAttributes`](/v2/docs/aws-cdk-lib.aws_kinesis.StreamConsumerAttributes) | Struct | A reference to a StreamConsumer, which can be imported using `StreamConsumer.fromStreamConsumerAttributes`. |
| [`StreamConsumerProps`](/v2/docs/aws-cdk-lib.aws_kinesis.StreamConsumerProps) | Struct | Properties for a Kinesis Stream Consumer. |
| [`StreamEncryption`](/v2/docs/aws-cdk-lib.aws_kinesis.StreamEncryption) | Enum | What kind of server-side encryption to apply to this stream. |
| [`StreamMode`](/v2/docs/aws-cdk-lib.aws_kinesis.StreamMode) | Enum | Specifies the capacity mode to apply to this stream. |
| [`StreamProps`](/v2/docs/aws-cdk-lib.aws_kinesis.StreamProps) | Struct | Properties for a Kinesis Stream. |