Amazon Kinesis Data Streams

Amazon AWS Data Stream

Synopsis

The Amazon Kinesis Data Streams device (type: awskinesis) reads a Kinesis data stream. Director lists the stream's shards, reads each of them, keeps a checkpoint for each, and writes every record.

Kinesis Data Streams is where AWS services and applications put records that more than one consumer reads: CloudWatch Logs through a subscription filter (CloudTrail, VPC Flow Logs, EKS audit logs, Route 53 Resolver query logs), EventBridge rules that forward GuardDuty and Security Hub findings, DynamoDB change streams, and what applications put. With this device Director is one of the stream's consumers. It needs no bucket, no queue and no DynamoDB table: the checkpoints are in Director's own state store.

Director calls AWS, and nothing reaches it from outside. It needs credentials that may read the stream — see Credentials and Permissions — and no public address, certificate or open port.

A record is written as it was put, and the device's pipeline parses it. The records of a CloudWatch Logs subscription are the exception: the device takes them apart into their log events — see CloudWatch Logs.

A stream that is delivered through Amazon Data Firehose's HTTP endpoint destination needs the Amazon Data Firehose device instead: Firehose pushes to Director, where this device pulls.

note

The Amazon Kinesis Data Streams device needs Director 2.3.1 or later.

Schema

- id: <numeric>
  name: <string>
  description: <string>
  type: awskinesis
  tags: <string[]>
  pipelines: <pipeline[]>
  status: <boolean>
  properties:
    stream: <string>
    region: <string>
    access_key_id: <string>
    secret_access_key: <string>
    session_token: <string>
    role_arn: <string>
    external_id: <string>
    endpoint: <string>
    start_position: <string>
    start_time: <string>
    poll_interval: <numeric>
    max_records: <numeric>
    content_encoding: <string>
    record_metadata: <boolean>
    timezone: <string>

Configuration

The following fields are used to define the device:

Device

Field	Required	Default	Description
`id`	Y	-	Unique numeric identifier
`name`	Y	-	Device name
`description`	N	-	Optional description
`type`	Y	-	Must be `awskinesis`
`tags`	N	-	Optional tags
`pipelines`	N	-	Optional pre-processor pipelines
`status`	N	`true`	Enable/disable the device

Stream

Field	Required	Default	Description
`stream`	Y	-	The stream's name, or its ARN (`arn:aws:kinesis:<region>:<account>:stream/<name>`). A stream of another account is named by its ARN. A name is up to 128 letters, digits, underscores, dots and hyphens
`region`	Y*	-	The stream's AWS region, such as `eu-west-1`
`endpoint`	N	-	An `http` or `https` URL that stands in for the service's, such as a VPC endpoint of Kinesis. STS is called through it too, so a VPC endpoint of Kinesis is used with keys and without `role_arn`: it does not answer STS, and the role cannot be assumed through it. A stand-in that answers both, such as LocalStack, takes a role

* = Required when stream is a name. With an ARN it may be left out, and when it is given it has to be the ARN's.

Credentials

Field	Required	Default	Description
`access_key_id`	N	-	With `secret_access_key`: both, or neither. Without them, Director's own environment answers — see Credentials and Permissions
`secret_access_key`	N	-	A secret: name an environment variable or a vault secret
`session_token`	N	-	The token of temporary keys. It goes with a key pair
`role_arn`	N	-	An IAM role to assume, with the keys above or with Director's own credentials
`external_id`	N	-	Sent when the role is assumed, up to 1224 characters. It goes with `role_arn`

Reading

Field	Required	Default	Description
`start_position`	N	`latest`	Where a shard is read from when the device reads the stream for the first time: `latest`, `trim_horizon` or `at_timestamp`. See Where the First Read Starts
`start_time`	Y*	-	The time for `at_timestamp`, in RFC 3339 (`2026-10-01T00:00:00Z`)
`content_encoding`	N	`auto`	How a record's data is decoded: `auto`, `gzip`, `zlib`, `deflate`, `zstd` or `none` (`identity` is taken for `none`). See Records
`record_metadata`	N	`false`	Write every event as a JSON object of the event (`data`) and what the stream says of its record: the stream, the shard, the sequence number, the partition key, and when it arrived. See Record Metadata

* = Required when start_position is at_timestamp, and refused with another position, where it would say something that is not used.

Advanced Configuration

Field	Required	Default	Description
`poll_interval`	N	`1`	Seconds a shard that has caught up waits before it is read again, from `1` to `300`. A shard that is behind is read again at once
`max_records`	N	`10000`	The most records one read asks for, from `1` to `10000`
`timezone`	N	-	The time zone written with the device's records

A configuration that cannot run stops the device with Configuration error and says why: no stream, or one that is neither a name nor an ARN; no region with a name, or a region that is not the ARN's; half a key pair, a session token without one, or an external ID without a role; a role that is not an IAM role's ARN; a start time without at_timestamp, or at_timestamp without one; a poll interval, a record count or a content encoding out of its bounds; or an endpoint that is not a URL. Every bound is checked when the configuration is read, before AWS is called.

Any change to these restarts the device's listener, a rotated key among them: the key is compared by its value. A secret that cannot be read when the configuration is compared, such as a vault that does not answer, changes nothing: the device goes on with the credentials it read.

Credentials and Permissions

The device signs in the way the Amazon S3 device does:

Keys in the configuration: access_key_id and secret_access_key, with session_token for temporary ones.
No keys: the AWS SDK's own chain, which is Director's environment — its AWS_* variables, the role of the EC2 instance or the ECS task it runs on, or a web identity on EKS.
A role (role_arn) on top of either: what 1 or 2 gives is used to assume it. This is how a stream of another account is read. The role's trust policy has to admit the caller, with the external ID when one is set, and the caller's own policy has to allow sts:AssumeRole on the role.

The credentials that read the stream — the role's, when one is assumed — need this statement in their policy:

{
  "Effect": "Allow",
  "Action": ["kinesis:ListShards", "kinesis:GetShardIterator", "kinesis:GetRecords"],
  "Resource": "arn:aws:kinesis:eu-west-1:123456789012:stream/cloudtrail-events"
}

and kms:Decrypt on the key when the stream is encrypted with a key of the account's own.

Device States

When the device starts it lists the stream's shards once, before it reads anything. What the stream answers decides the device's state:

The stream answers	State	What to check
Its shards	Connected
The stream is not there	Configuration error	The stream's name and its region
Access denied, an unknown key, a signature that does not match, a token that expired	Authentication error	The keys, the role, and the three permissions above
Anything else, such as no route or a timeout	Connection error	The network, a proxy, the `endpoint`

A device that cannot read its stream stops, and is started again after a wait that grows. A state store that does not answer when the device starts stops it with Initialization error: the device does not take that for a stream it never read, which would write over the checkpoints the store holds.

Records

content_encoding decides how a record's data is decoded:

`content_encoding`	A record becomes
`auto`	One event, without the line end its producer closed it with. A gzip, zstd or zlib record is decompressed first, each told by its header. A record that is the payload of a CloudWatch Logs subscription becomes its log events
`gzip`, `zlib`, `deflate`, `zstd`	One event, decompressed. A record that is not such data is written as it was put, with a warning
`none`	One event, exactly as it was put

deflate is raw deflate, with no zlib header (as .NET's DeflateStream writes); auto never takes a record for it. A zstd record may hold several frames.

An empty record is no event. A record the Kinesis Producer Library aggregated is taken apart into its user records, whatever the encoding, and each user record is then decoded as a record is.

Every event's request (_vmetric.event.request) is the stream's name. Every event is dated when its record arrived at the stream, apart from a CloudWatch log event.

A record that decompresses to more than 64 MiB is written as it was put, with a warning. An event larger than the device's buffer can ever admit (max_buffer_size, or a shard's share of it) is not written: the device logs a warning with its size and reads on.

CloudWatch Logs

A subscription filter puts a gzip-compressed JSON document per batch: the log group, the log stream and the log events. With content_encoding: auto each log event is written as one event:

{
  "id": "38378212459134562177402895230137412345678901234567890123",
  "timestamp": 1790000000123,
  "message": "START RequestId: 6f1e…",
  "logGroup": "/aws/lambda/orders",
  "logStream": "2026/10/03/[$LATEST]abc",
  "owner": "123456789012",
  "subscriptionFilters": ["to-kinesis"]
}

The field names are CloudWatch's own.
The event is dated by CloudWatch (timestamp, to the millisecond), not by when its record arrived.
message is the log line as CloudWatch holds it. A line that is JSON is JSON inside a string: the pipeline's json processor on message unfolds it.
A control message, which CloudWatch puts to check that the destination can be reached, is written as nothing, and the shard's checkpoint moves past it.

An object that has a subscription's message type and no list of log events is an application's own, and is written as it came.

Record Metadata

With record_metadata: true every event is written as a JSON object: what the device writes without the option is its data, and what the stream says of the event's record is beside it. It is off by default.

{
  "data": "user=ada action=login",
  "stream": "cloudtrail-events",
  "shard_id": "shardId-000000000001",
  "sequence_number": "49590338271490256608559692538361571095921575989136588898",
  "partition_key": "user-42",
  "approximate_arrival_timestamp": "2026-10-03T12:00:00.123Z"
}

Field	What it is
`data`	The event as the device writes it without the option: the record, decoded as `content_encoding` says, or a CloudWatch log event's object. Text when it is UTF-8, and base64 when it is not. A record that is JSON is JSON inside a string: the pipeline's `json` processor on `data` unfolds it
`data_encoding`	`base64` when `data` is base64. Left out for text
`stream`	The stream's name, which is the event's request too, also for a stream named by its ARN
`shard_id`	The shard the record was read from
`sequence_number`	The record's sequence number in its shard
`sub_sequence_number`	For a user record the Kinesis Producer Library aggregated: its place in the aggregate, from `0`. Left out for a record that is not an aggregate. The user records of one aggregate share its `sequence_number`
`partition_key`	The record's partition key; for a user record of an aggregate, its own
`approximate_arrival_timestamp`	When the record arrived at the stream, in RFC 3339, UTC, to the millisecond

The object is one line.
An event is dated, and has its request, as it is without the option: a CloudWatch log event by CloudWatch, any other when its record arrived.

Shards and Checkpoints

Every shard has a reader of its own. The device lists the stream's shards when it starts, every 30 seconds, and whenever a shard is read to its end. A shard is read at most four times a second while it is behind, and every poll_interval once it has caught up. A read the stream throttles is waited out: a shard gives five reads and 2 MB a second to all of its readers together. Any other failure is tried again after a wait that doubles up to 30 seconds, and is logged once for as long as it lasts, with a line when the shard is read again.

A child shard is read after its parents. A split closes a shard and makes two, and a merge closes two and makes one. The device reads a child only when every parent the stream still lists is read to its end, so the records of a partition key are written in the order they were put, across a split and a merge.

A checkpoint per shard, in Director's state store. A shard's checkpoint is the sequence number of the last record whose events were delivered: it moves when the delivery of what was written before it was confirmed, not when it was read. Delivery is at least once:

A restart, or another node of a cluster taking the device over, reads on after the last checkpoint that was written.
A Director that is stopped in order writes its checkpoints as it stops. What it had confirmed is not read twice.
A device that is restarted in place, by an edit that restarts it, reads again what was written and not yet confirmed. Those records arrive twice.
A Director that stops abruptly, by a crash or a kill, may not have written the checkpoints of its last second, and reads that second again.
What was written and not yet delivered when Director stopped is read again.
A shard that was read to its end is marked so, and is not read again.

The checkpoints are written at most once a second, and at once when the device stops. A write the state store refuses is tried again every second until it is taken, and is logged once for as long as the store refuses, with a line when the checkpoints are stored again. They are kept for a year, and a shard's is forgotten when the stream no longer lists the shard.

The checkpoints are stored under the device's ID and are of one stream: a device that is pointed at another stream starts that stream anew.

A checkpoint the stream no longer takes, because its record is older than the stream's retention, is replaced by the oldest record the shard holds, with a warning: the records in between are gone.

Run the device in memory mode — see Persistent Storage. The stream is the device's durable buffer: with crash_resistance, both Director's queue and the stream would be recovered from after a crash, and what was in the queue would be delivered twice.

One Director reads each device. A shard is read by one reader, with one set of checkpoints. The shards of a stream are not spread over the nodes of a cluster.

Where the First Read Starts

start_position is decided once, when the device first reads the stream, and is stored with the checkpoints:

`start_position`	A shard with no checkpoint is read from
`latest`	The moment the device first read the stream
`trim_horizon`	The oldest record the stream still holds
`at_timestamp`	`start_time`

A shard that had nothing delivered before a restart is read from that stored moment, not from the restart, so what was put while the device was down is read. A shard that a split or a merge makes later is read from its first record. To read a stream from another position, point a new device at it.

Console

In the console, go to Fleet Management > Devices and choose Amazon Kinesis Data Streams, under Amazon Web Services. The wizard's steps after General settings are:

Connection settings: Stream, Region, and Endpoint. Then Credentials: Access key ID, Secret access key, Session token, Role ARN, and External ID. Every credential is optional: leave them all empty to sign in with Director's own IAM role, fill the access keys to sign in with them, and the role to assume it with either. There is no authentication method to choose, as there is on the Amazon S3 device.
Read settings: Start position — Latest record, Oldest record, or At a time, with Start time — and Content encoding — Automatic, Gzip, Zlib, Deflate, Zstd, or None — and Record metadata (Disabled or Enabled, off by default).
Advanced settings: Poll interval, Maximum records, and the Simulation section.

On the device page, a stored Secret access key and Session token each have a Remove, which is how a stored key pair is taken off.

With no Director or cluster at version 2.3.1 or later selected, the wizard shows Amazon Kinesis Data Streams devices are unavailable on directors or clusters below version 2.3.1, and the device cannot be created.

Simulation

The device can be simulated as the other queue and stream devices are — see Device Simulation. A simulated device calls no AWS API, and its records carry the stream as their source.

Examples

Keys of the Same Account

Reading CloudTrail events that a CloudWatch Logs subscription puts to a stream, with a key pair whose secret is read from an environment variable...

devices:
  - id: 512400410
    name: cloudtrail-stream
    type: awskinesis
    status: true
    properties:
      region: "eu-west-1"
      stream: "cloudtrail-events"
      access_key_id: "AKIAIOSFODNN7EXAMPLE"
      secret_access_key: ${AWS_KINESIS_SECRET_ACCESS_KEY}
      start_position: latest

A Stream of Another Account

Reading a stream of another account by its ARN, with Director's own IAM role assuming a role of that account, from the oldest record the stream holds...

devices:
  - id: 512400411
    name: security-findings-stream
    type: awskinesis
    status: true
    properties:
      stream: "arn:aws:kinesis:us-east-1:210987654321:stream/security-findings"
      role_arn: "arn:aws:iam::210987654321:role/kinesis-reader"
      external_id: ${AWS_KINESIS_EXTERNAL_ID}
      start_position: trim_horizon

Synopsis​

Schema​

Configuration​

Device​

Stream​

Credentials​

Reading​

Advanced Configuration​

Credentials and Permissions​

Device States​

Records​

CloudWatch Logs​

Record Metadata​

Shards and Checkpoints​

Where the First Read Starts​

Console​

Simulation​

Examples​

Keys of the Same Account​

A Stream of Another Account​