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.
Arguments
Names below are TypeScript (camelCase). In Python use snake_case (e.g. dataClassification → data_classification); in Go use PascalCase on the Args struct (e.g. DataClassification).
provisioned object (KinesisProvisionedArgs) Switches the stream from on-demand to a fixed shardCount (required), e.g. { shardCount: 4 }. Omit for on-demand, which scales shards with write traffic and is the right choice unless throughput is steady and predictable.
retentionHours integer How long records stay readable, 24–8760 hours. Default: 24 (beyond that is billed per shard-hour). It is also how long a broken consumer has to recover: records it hasn't read by then expire and never reach its dead-letter queue. For DSQL CDC, 168 (7 days) is a safer choice.
maxRecordSizeKib integer Largest record the stream accepts, 1024–10240 KiB. Default: 2048 rather than AWS's 1024, because a DSQL row can be up to 2 MiB and a CDC event that doesn't fit fails delivery.
consumers array (KinesisConsumerArgs) Lambda functions that read the stream. Each field is explained under consumers below.
consumers 8 fields
For each consumer Anvil creates an event source mapping, a dead-letter queue (encrypted, keeps failures 14 days) and an IAM policy on the Lambda's role. By default the mapping starts from the oldest record so no CDC changes are missed. It splits a failing batch to isolate the bad record, and retries it at most 3 times instead of AWS's default of until it expires. Each consumer reads every record independently; about two share a stream's read throughput comfortably. Delivery is at-least-once and unordered, so handlers must tolerate repeats.
consumerName requiredstring Names the consumer's resources (<stream>-<consumerName>-dlq, -esm, -policy), so adding, removing or reordering consumers only touches the one that changed. Usually the Lambda's name. Letters, numbers and hyphens, starting with a letter, up to 32 characters, unique per stream. It is written by hand because a Lambda's ARN doesn't exist yet when Pulumi previews. Renaming it recreates the consumer's resources, including its dead-letter queue.
lambda requiredobject { arn, lambdaRoleArn }: pass fn.arn and fn.roleArn from an Anvil Lambda. The role ARN is needed to attach the consumer's permissions.
grantRead requiredboolean Grants the consumer's role the same permissions as stream.grantRead(fn), on this stream only. Required so the access is explicit, like every Anvil grant. It lives here rather than in a separate grantRead call because the event source mapping checks it can read the stream when it is created, so the grant has to exist first. false skips it with a deploy warning; the role then needs read access some other way. The role always gets sqs:SendMessage on its own dead-letter queue.
batchSize integer Most records sent to the Lambda per invocation, 1–10000. Default: 100.
reportBatchItemFailures boolean Lets the handler report which records in a batch failed, so only those are retried instead of the whole batch. The handler must return { batchItemFailures: [{ itemIdentifier }] }, so it is off by default. Default: false.
filter object Only invokes the Lambda for matching records; the rest are skipped and not billed. DSQL CDC sends every table to the stream and can't filter itself, so this is where per-consumer filtering happens. tables matches the CDC event's table: "orders" in any schema, "app.orders" only in schema app. patterns takes raw Lambda filter patterns (JSON) for non-CDC data. A record matching any entry is delivered; at most 5 patterns per consumer (unqualified tables share one). A changed filter takes a minute or two to apply, so records arriving right after a deploy may still use the old one.
startingPosition string Where the consumer starts reading when it is first created. Changing it later recreates the event source mapping, which resets the consumer's position. Default: "TRIM_HORIZON".
TRIM_HORIZON— Every record still in the stream, so a consumer that builds state (search index, replica, read model) misses nothing from before it existed. Default.LATEST— Only records written after the consumer is created, for consumers that only care about what happens from now on (cache invalidation, notifications). Skips any backlog when adding a consumer to a stream that has been running.
enabled boolean Set to false to pause the consumer without deleting it. It keeps its position and resumes where it left off, but records older than retentionHours expire while it is paused. Default: true.
This component also accepts a hatch argument to override the underlying resource
configuration. See Overrides & Escape Hatch.
Attributes
Outputs available once the resource is deployed.
arn string The ARN of the Kinesis stream. Pass it to DSQL as the CDC target.
name string The physical name of the Kinesis stream.
dlqArns map<string, string> Each consumer's consumerName mapped to the ARN of its dead-letter queue.
dlqUrls map<string, string> Each consumer's consumerName mapped to the URL of its dead-letter queue.