KinesisStream

An encrypted, on-demand Kinesis Data Stream sized for Aurora DSQL change data capture. Wire Lambda consumers that each get a dead-letter queue, safe retries and an explicit read grant.

Messaging · anvil:aws:KinesisStream

Always encrypted with the AWS-managed key alias/aws/kinesis, which is free and needs no KMS permissions; your own KMS key is opt-in through hatch.kinesis. Protected from deletion by default, since deleting a stream deletes every record no consumer has read yet.

Minimal

The smallest config that deploys.

import { App } from '@anvil-cloud/sdk';
import * as anvil from '@anvil-cloud/sdk';

export default new App({
  run(ctx) {
    const events = new anvil.aws.KinesisStream('events');
  },
});

import anvil_cloud as anvil

def infra(ctx: anvil.Context):
    events = anvil.aws.KinesisStream("events")

anvil.App(run=infra)

package main

import (
    "github.com/anvil-cloud/anvil/sdk/go/anvil"
    anvilaws "github.com/anvil-cloud/anvil/sdk/go/anvil/aws"
)

func main() {
    anvil.Run(anvil.AppConfig{
        Run: func(ctx *anvil.Context) error {
            _, err := anvilaws.NewKinesisStream(ctx.PulumiCtx(), "events", nil)
            if err != nil {
                return err
            }
            return nil
        },
    })
}

Granting access

Lambdas that process records are consumers, each with grantRead: true. Compute that writes records gets grantWrite.

const events = new anvil.aws.KinesisStream('events', {
  consumers: [{
    consumerName: 'indexer',
    lambda: { arn: indexer.arn, lambdaRoleArn: indexer.roleArn },
    grantRead: true,
  }],
});

events.grantWrite(api);

events = anvil.aws.KinesisStream("events",
    consumers=[{
        "consumer_name": "indexer",
        # "lambda_" (trailing underscore) because "lambda" is a Python keyword.
        "lambda_": {"arn": indexer.arn, "lambda_role_arn": indexer.role_arn},
        "grant_read": True,
    }],
)

events.grant_write(api)

events, err := anvilaws.NewKinesisStream(ctx.PulumiCtx(), "events", &anvilaws.KinesisStreamArgs{
    Consumers: anvilaws.KinesisConsumerArgsArray{
        &anvilaws.KinesisConsumerArgsArgs{
            ConsumerName: "indexer",
            Lambda:       &anvilaws.KinesisLambdaConsumerArgsArgs{Arn: indexer.Arn, LambdaRoleArn: indexer.RoleArn},
            GrantRead:    pulumi.Bool(true),
        },
    },
})
if err != nil {
    return err
}

// Go has no grant methods yet: give the Lambda the same actions with permissions.
api, err := anvilaws.NewLambda(ctx.PulumiCtx(), "api", &anvilaws.LambdaArgs{
    Runtime: anvilaws.LambdaRuntime_Nodejs24_x,
    Handler: pulumi.String("functions/api/index.handler"),
    Permissions: anvilaws.LambdaPermissionArray{
        anvilaws.LambdaPermissionArgs{
            Actions:  pulumi.ToStringArray([]string{"kinesis:PutRecord", "kinesis:PutRecords", "kinesis:DescribeStreamSummary", "kinesis:ListShards"}),
            Resource: events.Arn,
        },
    },
})
if err != nil {
    return err
}

What Anvil creates

grantRead (and a consumer's grantRead: true) adds GetRecords, GetShardIterator, DescribeStream, DescribeStreamSummary and ListShards on this stream only.
grantWrite adds PutRecord, PutRecords, DescribeStreamSummary and ListShards. grantFullAccess is both.
Each consumer also gets sqs:SendMessage on its own dead-letter queue, and nothing on any other consumer's.
Each grant is an inline policy on the compute's role, removed when you remove the grant.

Good to know

Records stay readable for 24 hours by default. That's also how long a broken consumer has to recover, so consider retentionHours: 168.
Delivery to consumers is at least once and unordered, so handlers must tolerate repeats.
A failing batch is split to isolate the bad record and retried at most 3 times, then the record goes to that consumer's dead-letter queue.
Deletion protection is on: deleting a stream deletes every record no consumer has read yet.

The Full example tab has a complete program. Every argument and attribute is on the Reference tab.