aws-cdk-lib.aws_kinesis

Amazon Kinesis Construct Library

Amazon Kinesis provides collection and processing of large streams 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.

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

kinesis.Stream(self, "MyFirstStream",
    stream_name="my-awesome-stream"
)

Stream.Builder.create(this, "MyFirstStream")
        .streamName("my-awesome-stream")
        .build();

new Stream(this, "MyFirstStream", new StreamProps {
    StreamName = "my-awesome-stream"
});

kinesis.NewStream(this, jsii.String("MyFirstStream"), &StreamProps{
	StreamName: jsii.String("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

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

kinesis.Stream(self, "MyFirstStream",
    stream_name="my-awesome-stream",
    shard_count=3,
    retention_period=Duration.hours(48)
)

Stream.Builder.create(this, "MyFirstStream")
        .streamName("my-awesome-stream")
        .shardCount(3)
        .retentionPeriod(Duration.hours(48))
        .build();

new Stream(this, "MyFirstStream", new StreamProps {
    StreamName = "my-awesome-stream",
    ShardCount = 3,
    RetentionPeriod = Duration.Hours(48)
});

kinesis.NewStream(this, jsii.String("MyFirstStream"), &StreamProps{
	StreamName: jsii.String("my-awesome-stream"),
	ShardCount: jsii.Number(3),
	RetentionPeriod: awscdk.Duration_Hours(jsii.Number(48)),
})

Encryption

Stream encryption 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.

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

kinesis.Stream(self, "MyEncryptedStream")

new Stream(this, "MyEncryptedStream");

new Stream(this, "MyEncryptedStream");

kinesis.NewStream(this, jsii.String("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.

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

kinesis.Stream(self, "MyEncryptedStream",
    encryption=kinesis.StreamEncryption.KMS
)

Stream.Builder.create(this, "MyEncryptedStream")
        .encryption(StreamEncryption.KMS)
        .build();

new Stream(this, "MyEncryptedStream", new StreamProps {
    Encryption = StreamEncryption.KMS
});

kinesis.NewStream(this, jsii.String("MyEncryptedStream"), &StreamProps{
	Encryption: kinesis.StreamEncryption_KMS,
})

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

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

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

key = kms.Key(self, "MyKey")

kinesis.Stream(self, "MyEncryptedStream",
    encryption=kinesis.StreamEncryption.KMS,
    encryption_key=key
)

Key key = new Key(this, "MyKey");

Stream.Builder.create(this, "MyEncryptedStream")
        .encryption(StreamEncryption.KMS)
        .encryptionKey(key)
        .build();

var key = new Key(this, "MyKey");

new Stream(this, "MyEncryptedStream", new StreamProps {
    Encryption = StreamEncryption.KMS,
    EncryptionKey = key
});

key := kms.NewKey(this, jsii.String("MyKey"))

kinesis.NewStream(this, jsii.String("MyEncryptedStream"), &StreamProps{
	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

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

imported_stream = kinesis.Stream.from_stream_arn(self, "ImportedStream", "arn:aws:kinesis:us-east-2:123456789012:stream/f3j09j2230j")

IStream importedStream = Stream.fromStreamArn(this, "ImportedStream", "arn:aws:kinesis:us-east-2:123456789012:stream/f3j09j2230j");

var importedStream = Stream.FromStreamArn(this, "ImportedStream", "arn:aws:kinesis:us-east-2:123456789012:stream/f3j09j2230j");

importedStream := kinesis.Stream_FromStreamArn(this, jsii.String("ImportedStream"), jsii.String("arn:aws:kinesis:us-east-2:123456789012:stream/f3j09j2230j"))

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

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',
  ),
});

imported_stream = kinesis.Stream.from_stream_attributes(self, "ImportedEncryptedStream",
    stream_arn="arn:aws:kinesis:us-east-2:123456789012:stream/f3j09j2230j",
    encryption_key=kms.Key.from_key_arn(self, "key", "arn:aws:kms:us-east-1:123456789012:key/12345678-1234-1234-1234-123456789012")
)

IStream importedStream = Stream.fromStreamAttributes(this, "ImportedEncryptedStream", StreamAttributes.builder()
        .streamArn("arn:aws:kinesis:us-east-2:123456789012:stream/f3j09j2230j")
        .encryptionKey(Key.fromKeyArn(this, "key", "arn:aws:kms:us-east-1:123456789012:key/12345678-1234-1234-1234-123456789012"))
        .build());

var importedStream = Stream.FromStreamAttributes(this, "ImportedEncryptedStream", new StreamAttributes {
    StreamArn = "arn:aws:kinesis:us-east-2:123456789012:stream/f3j09j2230j",
    EncryptionKey = Key.FromKeyArn(this, "key", "arn:aws:kms:us-east-1:123456789012:key/12345678-1234-1234-1234-123456789012")
});

importedStream := kinesis.Stream_FromStreamAttributes(this, jsii.String("ImportedEncryptedStream"), &StreamAttributes{
	StreamArn: jsii.String("arn:aws:kinesis:us-east-2:123456789012:stream/f3j09j2230j"),
	EncryptionKey: kms.Key_FromKeyArn(this, jsii.String("key"), jsii.String("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.

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);

lambda_role = iam.Role(self, "Role",
    assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
    description="Example role..."
)

stream = kinesis.Stream(self, "MyEncryptedStream",
    encryption=kinesis.StreamEncryption.KMS
)

# give lambda permissions to read stream
stream.grant_read(lambda_role)

Role lambdaRole = Role.Builder.create(this, "Role")
        .assumedBy(new ServicePrincipal("lambda.amazonaws.com"))
        .description("Example role...")
        .build();

Stream stream = Stream.Builder.create(this, "MyEncryptedStream")
        .encryption(StreamEncryption.KMS)
        .build();

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

var lambdaRole = new Role(this, "Role", new RoleProps {
    AssumedBy = new ServicePrincipal("lambda.amazonaws.com"),
    Description = "Example role..."
});

var stream = new Stream(this, "MyEncryptedStream", new StreamProps {
    Encryption = StreamEncryption.KMS
});

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

lambdaRole := iam.NewRole(this, jsii.String("Role"), &RoleProps{
	AssumedBy: iam.NewServicePrincipal(jsii.String("lambda.amazonaws.com")),
	Description: jsii.String("Example role..."),
})

stream := kinesis.NewStream(this, jsii.String("MyEncryptedStream"), &StreamProps{
	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.

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);

lambda_role = iam.Role(self, "Role",
    assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
    description="Example role..."
)

stream = kinesis.Stream(self, "MyEncryptedStream",
    encryption=kinesis.StreamEncryption.KMS
)

# give lambda permissions to write to stream
stream.grant_write(lambda_role)

Role lambdaRole = Role.Builder.create(this, "Role")
        .assumedBy(new ServicePrincipal("lambda.amazonaws.com"))
        .description("Example role...")
        .build();

Stream stream = Stream.Builder.create(this, "MyEncryptedStream")
        .encryption(StreamEncryption.KMS)
        .build();

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

var lambdaRole = new Role(this, "Role", new RoleProps {
    AssumedBy = new ServicePrincipal("lambda.amazonaws.com"),
    Description = "Example role..."
});

var stream = new Stream(this, "MyEncryptedStream", new StreamProps {
    Encryption = StreamEncryption.KMS
});

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

lambdaRole := iam.NewRole(this, jsii.String("Role"), &RoleProps{
	AssumedBy: iam.NewServicePrincipal(jsii.String("lambda.amazonaws.com")),
	Description: jsii.String("Example role..."),
})

stream := kinesis.NewStream(this, jsii.String("MyEncryptedStream"), &StreamProps{
	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.

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');

user = iam.User(self, "MyUser")

stream = kinesis.Stream(self, "MyStream")

# give my user permissions to list shards
stream.grant(user, "kinesis:ListShards")

User user = new User(this, "MyUser");

Stream stream = new Stream(this, "MyStream");

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

var user = new User(this, "MyUser");

var stream = new Stream(this, "MyStream");

// give my user permissions to list shards
stream.Grant(user, "kinesis:ListShards");

user := iam.NewUser(this, jsii.String("MyUser"))

stream := kinesis.NewStream(this, jsii.String("MyStream"))

// give my user permissions to list shards
stream.grant(user, jsii.String("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.

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' });

stream = kinesis.Stream(self, "MyStream")

# Using base metric method passing the metric name
stream.metric("GetRecords.Success")

# using pre-defined metric method
stream.metric_get_records_success()

# using pre-defined and overriding the statistic
stream.metric_get_records_success(statistic="Maximum")

Stream stream = new 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(MetricOptions.builder().statistic("Maximum").build());

var stream = new 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(new MetricOptions { Statistic = "Maximum" });

stream := kinesis.NewStream(this, jsii.String("MyStream"))

// Using base metric method passing the metric name
stream.metric(jsii.String("GetRecords.Success"))

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

// using pre-defined and overriding the statistic
stream.metricGetRecordsSuccess(&MetricOptions{
	Statistic: jsii.String("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.

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

stream = kinesis.Stream(self, "MyStream",
    shard_level_metrics=[kinesis.ShardLevelMetrics.ALL]
)

Stream stream = Stream.Builder.create(this, "MyStream")
        .shardLevelMetrics(List.of(ShardLevelMetrics.ALL))
        .build();

var stream = new Stream(this, "MyStream", new StreamProps {
    ShardLevelMetrics = new [] { ShardLevelMetrics.ALL }
});

stream := kinesis.NewStream(this, jsii.String("MyStream"), &StreamProps{
	ShardLevelMetrics: []ShardLevelMetrics{
		kinesis.ShardLevelMetrics_ALL,
	},
})

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

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,
  ],
});

stream = kinesis.Stream(self, "MyStream",
    shard_level_metrics=[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
    ]
)

Stream stream = Stream.Builder.create(this, "MyStream")
        .shardLevelMetrics(List.of(ShardLevelMetrics.INCOMING_BYTES, ShardLevelMetrics.INCOMING_RECORDS, ShardLevelMetrics.ITERATOR_AGE_MILLISECONDS, ShardLevelMetrics.OUTGOING_BYTES, ShardLevelMetrics.OUTGOING_RECORDS, ShardLevelMetrics.READ_PROVISIONED_THROUGHPUT_EXCEEDED, ShardLevelMetrics.WRITE_PROVISIONED_THROUGHPUT_EXCEEDED))
        .build();

var stream = new Stream(this, "MyStream", new StreamProps {
    ShardLevelMetrics = new [] { ShardLevelMetrics.INCOMING_BYTES, ShardLevelMetrics.INCOMING_RECORDS, ShardLevelMetrics.ITERATOR_AGE_MILLISECONDS, ShardLevelMetrics.OUTGOING_BYTES, ShardLevelMetrics.OUTGOING_RECORDS, ShardLevelMetrics.READ_PROVISIONED_THROUGHPUT_EXCEEDED, ShardLevelMetrics.WRITE_PROVISIONED_THROUGHPUT_EXCEEDED }
});

stream := kinesis.NewStream(this, jsii.String("MyStream"), &StreamProps{
	ShardLevelMetrics: []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.

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.

To create and associate a stream consumer with a stream

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

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

stream = kinesis.Stream(self, "MyStream")

stream_consumer = kinesis.StreamConsumer(self, "MyStreamConsumer",
    stream_consumer_name="MyStreamConsumer",
    stream=stream
)

Stream stream = new Stream(this, "MyStream");

StreamConsumer streamConsumer = StreamConsumer.Builder.create(this, "MyStreamConsumer")
        .streamConsumerName("MyStreamConsumer")
        .stream(stream)
        .build();

var stream = new Stream(this, "MyStream");

var streamConsumer = new StreamConsumer(this, "MyStreamConsumer", new StreamConsumerProps {
    StreamConsumerName = "MyStreamConsumer",
    Stream = stream
});

stream := kinesis.NewStream(this, jsii.String("MyStream"))

streamConsumer := kinesis.NewStreamConsumer(this, jsii.String("MyStreamConsumer"), &StreamConsumerProps{
	StreamConsumerName: jsii.String("MyStreamConsumer"),
	Stream: Stream,
})

Read Permissions

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

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);

lambda_role = iam.Role(self, "Role",
    assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
    description="Example role..."
)

stream = kinesis.Stream(self, "MyEncryptedStream",
    encryption=kinesis.StreamEncryption.KMS
)
stream_consumer = kinesis.StreamConsumer(self, "MyStreamConsumer",
    stream_consumer_name="MyStreamConsumer",
    stream=stream
)

# give lambda permissions to read stream via the stream consumer
stream_consumer.grant_read(lambda_role)

Role lambdaRole = Role.Builder.create(this, "Role")
        .assumedBy(new ServicePrincipal("lambda.amazonaws.com"))
        .description("Example role...")
        .build();

Stream stream = Stream.Builder.create(this, "MyEncryptedStream")
        .encryption(StreamEncryption.KMS)
        .build();
StreamConsumer streamConsumer = StreamConsumer.Builder.create(this, "MyStreamConsumer")
        .streamConsumerName("MyStreamConsumer")
        .stream(stream)
        .build();

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

var lambdaRole = new Role(this, "Role", new RoleProps {
    AssumedBy = new ServicePrincipal("lambda.amazonaws.com"),
    Description = "Example role..."
});

var stream = new Stream(this, "MyEncryptedStream", new StreamProps {
    Encryption = StreamEncryption.KMS
});
var streamConsumer = new StreamConsumer(this, "MyStreamConsumer", new StreamConsumerProps {
    StreamConsumerName = "MyStreamConsumer",
    Stream = stream
});

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

lambdaRole := iam.NewRole(this, jsii.String("Role"), &RoleProps{
	AssumedBy: iam.NewServicePrincipal(jsii.String("lambda.amazonaws.com")),
	Description: jsii.String("Example role..."),
})

stream := kinesis.NewStream(this, jsii.String("MyEncryptedStream"), &StreamProps{
	Encryption: kinesis.StreamEncryption_KMS,
})
streamConsumer := kinesis.NewStreamConsumer(this, jsii.String("MyStreamConsumer"), &StreamConsumerProps{
	StreamConsumerName: jsii.String("MyStreamConsumer"),
	Stream: 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.

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:

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()],
}));

stream = kinesis.Stream(self, "MyStream")
stream_consumer = kinesis.StreamConsumer(self, "MyStreamConsumer",
    stream_consumer_name="MyStreamConsumer",
    stream=stream
)

# create a stream resource policy via addToResourcePolicy method
stream.add_to_resource_policy(iam.PolicyStatement(
    resources=[stream.stream_arn],
    actions=["kinesis:GetRecords"],
    principals=[iam.AnyPrincipal()]
))

# create a stream consumer resource policy via addToResourcePolicy method
stream_consumer.add_to_resource_policy(iam.PolicyStatement(
    resources=[stream.stream_arn],
    actions=["kinesis:DescribeStreamConsumer"],
    principals=[iam.AnyPrincipal()]
))

Stream stream = new Stream(this, "MyStream");
StreamConsumer streamConsumer = StreamConsumer.Builder.create(this, "MyStreamConsumer")
        .streamConsumerName("MyStreamConsumer")
        .stream(stream)
        .build();

// create a stream resource policy via addToResourcePolicy method
stream.addToResourcePolicy(PolicyStatement.Builder.create()
        .resources(List.of(stream.getStreamArn()))
        .actions(List.of("kinesis:GetRecords"))
        .principals(List.of(new AnyPrincipal()))
        .build());

// create a stream consumer resource policy via addToResourcePolicy method
streamConsumer.addToResourcePolicy(PolicyStatement.Builder.create()
        .resources(List.of(stream.getStreamArn()))
        .actions(List.of("kinesis:DescribeStreamConsumer"))
        .principals(List.of(new AnyPrincipal()))
        .build());

var stream = new Stream(this, "MyStream");
var streamConsumer = new StreamConsumer(this, "MyStreamConsumer", new StreamConsumerProps {
    StreamConsumerName = "MyStreamConsumer",
    Stream = stream
});

// create a stream resource policy via addToResourcePolicy method
stream.AddToResourcePolicy(new PolicyStatement(new PolicyStatementProps {
    Resources = new [] { stream.StreamArn },
    Actions = new [] { "kinesis:GetRecords" },
    Principals = new [] { new AnyPrincipal() }
}));

// create a stream consumer resource policy via addToResourcePolicy method
streamConsumer.AddToResourcePolicy(new PolicyStatement(new PolicyStatementProps {
    Resources = new [] { stream.StreamArn },
    Actions = new [] { "kinesis:DescribeStreamConsumer" },
    Principals = new [] { new AnyPrincipal() }
}));

stream := kinesis.NewStream(this, jsii.String("MyStream"))
streamConsumer := kinesis.NewStreamConsumer(this, jsii.String("MyStreamConsumer"), &StreamConsumerProps{
	StreamConsumerName: jsii.String("MyStreamConsumer"),
	Stream: Stream,
})

// create a stream resource policy via addToResourcePolicy method
stream.addToResourcePolicy(iam.NewPolicyStatement(&PolicyStatementProps{
	Resources: []*string{
		stream.streamArn,
	},
	Actions: []*string{
		jsii.String("kinesis:GetRecords"),
	},
	Principals: []IPrincipal{
		iam.NewAnyPrincipal(),
	},
}))

// create a stream consumer resource policy via addToResourcePolicy method
streamConsumer.addToResourcePolicy(iam.NewPolicyStatement(&PolicyStatementProps{
	Resources: []*string{
		stream.streamArn,
	},
	Actions: []*string{
		jsii.String("kinesis:DescribeStreamConsumer"),
	},
	Principals: []IPrincipal{
		iam.NewAnyPrincipal(),
	},
}))

You can create a resource manually by using ResourcePolicy. Also, you can set a custom policy document to ResourcePolicy. If not, a blank policy document will be set.

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,
});

stream = kinesis.Stream(self, "MyStream")
stream_consumer = kinesis.StreamConsumer(self, "MyStreamConsumer",
    stream_consumer_name="MyStreamConsumer",
    stream=stream
)

# create a custom policy document
policy_document = iam.PolicyDocument(
    assign_sids=True,
    statements=[
        iam.PolicyStatement(
            actions=["kinesis:GetRecords"],
            resources=[stream.stream_arn],
            principals=[iam.AnyPrincipal()]
        )
    ]
)

# create a stream resource policy manually
kinesis.ResourcePolicy(self, "ResourcePolicy",
    stream=stream,
    policy_document=policy_document
)

# create a stream consumer resource policy manually
kinesis.ResourcePolicy(self, "ResourcePolicy",
    stream_consumer=stream_consumer,
    policy_document=policy_document
)

Stream stream = new Stream(this, "MyStream");
StreamConsumer streamConsumer = StreamConsumer.Builder.create(this, "MyStreamConsumer")
        .streamConsumerName("MyStreamConsumer")
        .stream(stream)
        .build();

// create a custom policy document
PolicyDocument policyDocument = PolicyDocument.Builder.create()
        .assignSids(true)
        .statements(List.of(
            PolicyStatement.Builder.create()
                    .actions(List.of("kinesis:GetRecords"))
                    .resources(List.of(stream.getStreamArn()))
                    .principals(List.of(new AnyPrincipal()))
                    .build()))
        .build();

// create a stream resource policy manually
// create a stream resource policy manually
ResourcePolicy.Builder.create(this, "ResourcePolicy")
        .stream(stream)
        .policyDocument(policyDocument)
        .build();

// create a stream consumer resource policy manually
// create a stream consumer resource policy manually
ResourcePolicy.Builder.create(this, "ResourcePolicy")
        .streamConsumer(streamConsumer)
        .policyDocument(policyDocument)
        .build();

var stream = new Stream(this, "MyStream");
var streamConsumer = new StreamConsumer(this, "MyStreamConsumer", new StreamConsumerProps {
    StreamConsumerName = "MyStreamConsumer",
    Stream = stream
});

// create a custom policy document
var policyDocument = new PolicyDocument(new PolicyDocumentProps {
    AssignSids = true,
    Statements = new [] {
        new PolicyStatement(new PolicyStatementProps {
            Actions = new [] { "kinesis:GetRecords" },
            Resources = new [] { stream.StreamArn },
            Principals = new [] { new AnyPrincipal() }
        }) }
});

// create a stream resource policy manually
// create a stream resource policy manually
new ResourcePolicy(this, "ResourcePolicy", new ResourcePolicyProps {
    Stream = stream,
    PolicyDocument = policyDocument
});

// create a stream consumer resource policy manually
// create a stream consumer resource policy manually
new ResourcePolicy(this, "ResourcePolicy", new ResourcePolicyProps {
    StreamConsumer = streamConsumer,
    PolicyDocument = policyDocument
});

stream := kinesis.NewStream(this, jsii.String("MyStream"))
streamConsumer := kinesis.NewStreamConsumer(this, jsii.String("MyStreamConsumer"), &StreamConsumerProps{
	StreamConsumerName: jsii.String("MyStreamConsumer"),
	Stream: Stream,
})

// create a custom policy document
policyDocument := iam.NewPolicyDocument(&PolicyDocumentProps{
	AssignSids: jsii.Boolean(true),
	Statements: []PolicyStatement{
		iam.NewPolicyStatement(&PolicyStatementProps{
			Actions: []*string{
				jsii.String("kinesis:GetRecords"),
			},
			Resources: []*string{
				stream.streamArn,
			},
			Principals: []IPrincipal{
				iam.NewAnyPrincipal(),
			},
		}),
	},
})

// create a stream resource policy manually
// create a stream resource policy manually
kinesis.NewResourcePolicy(this, jsii.String("ResourcePolicy"), &ResourcePolicyProps{
	Stream: Stream,
	PolicyDocument: PolicyDocument,
})

// create a stream consumer resource policy manually
// create a stream consumer resource policy manually
kinesis.NewResourcePolicy(this, jsii.String("ResourcePolicy"), &ResourcePolicyProps{
	StreamConsumer: StreamConsumer,
	PolicyDocument: PolicyDocument,
})

Types

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