Skip to main content

aws_kinesis input

Receive messages from one or more Kinesis streams.

# Common config fields, showing default values
input:
  label: ""
  aws_kinesis:
    streams: [] # No default (required)
    dynamodb:
      table: ""
      create: false
    checkpoint_limit: 1024
    auto_replay_nacks: true
    commit_period: "5s"
    steal_grace_period: "2s"
    start_from_oldest: true
    batching:
      count: 0
      byte_size: 0
      period: ""
      check: ""

Consumes messages from one or more Kinesis streams either by automatically balancing shards across other instances of this input, or by consuming shards listed explicitly. The latest message sequence consumed by this input is stored within a DynamoDB table, which allows it to resume at the correct sequence of the shard during restarts. This table is also used for coordination across distributed inputs when shard balancing.

Expanso Edge will not store a consumed sequence unless it is acknowledged at the output level, which ensures at-least-once delivery guarantees.

Ordering​

By default messages of a shard can be processed in parallel, up to a limit determined by the field checkpoint_limit. However, if strict ordered processing is required then this value must be set to 1 in order to process shard messages in lock-step. When doing so it is recommended that you perform batching at this component for performance as it will not be possible to batch lock-stepped messages at the output level.

Enhanced fan-out​

Kinesis enforces a shared limit of 5 GetRecords calls per second per shard across all polling consumers of a stream. When multiple applications consume the same stream this budget is quickly exhausted and consumers receive ReadProvisionedThroughputExceeded errors. There are two remedies:

  • Set the poll_period field to bound how frequently this input polls each shard, leaving headroom for other consumers.
  • Enable enhanced_fan_out, which registers this pipeline as a dedicated stream consumer with its own 2MB/s per shard read throughput, delivered over HTTP/2 push rather than polling. Enhanced fan-out requires the IAM permissions kinesis:DescribeStreamConsumer and kinesis:SubscribeToShard, and incurs additional AWS charges per consumer-shard-hour plus data retrieval. The named consumer is registered automatically on first use and is never deregistered; that registration additionally requires kinesis:RegisterStreamConsumer, which may be omitted when a consumer with the configured name already exists on every stream (for example one provisioned through infrastructure-as-code).

Table schema​

It's possible to configure Expanso Edge to create the DynamoDB table required for coordination if it does not already exist. However, if you wish to create this yourself (recommended) then create a table with a string HASH key StreamID and a string RANGE key ShardID.

Batching​

Use the batching fields to configure an optional batching policy. Each stream shard will be batched separately in order to ensure that acknowledgements aren't contaminated.

Fields​

streams​

One or more Kinesis data streams to consume from. Streams can either be specified by their name or full ARN. Shards of a stream are automatically balanced across consumers by coordinating through the provided DynamoDB table. Multiple comma separated streams can be listed in a single element. Shards are automatically distributed across consumers of a stream by coordinating through the provided DynamoDB table. Alternatively, it's possible to specify an explicit shard to consume from with a colon after the stream name, e.g. foo:0 would consume the shard 0 of the stream foo.

Type: array of string

dynamodb​

Determines the table used for storing and accessing the latest consumed sequence for shards, and for coordinating balanced consumers of streams.

Type: object

dynamodb.table​

The name of the table to access.

Type: string
Default: ""

dynamodb.create​

Whether, if the table does not exist, it should be created.

Type: bool
Default: false

dynamodb.billing_mode​

When creating the table determines the billing mode.

Type: string
Default: "PAY_PER_REQUEST"

Options: PROVISIONED, PAY_PER_REQUEST

dynamodb.read_capacity_units​

Set the provisioned read capacity when creating the table with a billing_mode of PROVISIONED.

Type: int
Default: 0

dynamodb.write_capacity_units​

Set the provisioned write capacity when creating the table with a billing_mode of PROVISIONED.

Type: int
Default: 0

dynamodb.region​

The AWS region to target.

Type: string

dynamodb.endpoint​

Allows you to specify a custom endpoint for the AWS API.

Type: string

dynamodb.tcp​

TCP socket configuration.

Type: object

dynamodb.tcp.connect_timeout​

Maximum amount of time a dial will wait for a connect to complete. Zero disables.

Type: string
Default: "0s"

dynamodb.tcp.keep_alive​

TCP keep-alive probe configuration.

Type: object

dynamodb.tcp.keep_alive.idle​

Duration the connection must be idle before sending the first keep-alive probe. Zero defaults to 15s. Negative values disable keep-alive probes.

Type: string
Default: "15s"

dynamodb.tcp.keep_alive.interval​

Duration between keep-alive probes. Zero defaults to 15s.

Type: string
Default: "15s"

dynamodb.tcp.keep_alive.count​

Maximum unanswered keep-alive probes before dropping the connection. Zero defaults to 9.

Type: int
Default: 9

dynamodb.tcp.tcp_user_timeout​

Maximum time to wait for acknowledgment of transmitted data before killing the connection. Linux-only (kernel 2.6.37+), ignored on other platforms. When enabled, keep_alive.idle must be greater than this value per RFC 5482. Zero disables.

Type: string
Default: "0s"

dynamodb.credentials​

Optional manual configuration of AWS credentials to use. More information can be found in this document.

Type: object

dynamodb.credentials.profile​

A profile from ~/.aws/credentials to use.

Type: string

dynamodb.credentials.id​

The ID of credentials to use.

Type: string

dynamodb.credentials.secret​

The secret for the credentials being used.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string

dynamodb.credentials.token​

The token for the credentials being used, required when using short term credentials.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string

dynamodb.credentials.from_ec2_role​

Use the credentials of a host EC2 machine configured to assume an IAM role associated with the instance.

Type: bool

dynamodb.credentials.role​

A role ARN to assume.

Type: string

dynamodb.credentials.role_external_id​

An external ID to provide when assuming a role.

Type: string

checkpoint_limit​

The maximum gap between the in flight sequence versus the latest acknowledged sequence at a given time. Increasing this limit enables parallel processing and batching at the output level to work on individual shards. Any given sequence will not be committed unless all messages under that offset are delivered in order to preserve at least once delivery guarantees.

Type: int
Default: 1024

poll_period​

An optional minimum period between GetRecords calls made against each shard. Kinesis allows a shared budget of 5 GetRecords calls per second per shard across all consumers of a stream, so setting this to e.g. 250ms bounds this consumer to roughly four reads per second per shard, leaving headroom for other consumers of the same stream. The default of 0s polls as fast as records are consumed. This setting has no effect when enhanced_fan_out is enabled. A shard is polled at most once per period, so the committed sequence advances no faster than that; values above lease_period are rejected.

Type: string
Default: "0s"

enhanced_fan_out​

Consume the stream using enhanced fan-out, which provides this consumer dedicated read throughput of 2MB/s per shard via HTTP/2 push delivery, avoiding the 5 reads per second per shard limit that polling consumers share. The named consumer is registered on each stream automatically if it does not already exist (and is never deregistered). Requires the IAM permissions kinesis:DescribeStreamConsumer and kinesis:SubscribeToShard, plus kinesis:RegisterStreamConsumer unless a consumer with the configured name already exists on every stream (for example one provisioned through infrastructure-as-code). Note that AWS bills enhanced fan-out consumers per consumer-shard-hour plus data retrieval. When a shard has no stored checkpoint and start_from_oldest is false, consumption starts from a timestamp anchored to this host's clock at claim time (rather than the server-side LATEST used by polling, which cannot be pinned across resubscribes without risking skipped records), so a clock running ahead can omit its offset's worth of records after a fresh claim.

Type: object

enhanced_fan_out.enabled​

Whether to consume the stream using enhanced fan-out.

Type: bool
Default: false

enhanced_fan_out.consumer_name​

The name of the enhanced fan-out consumer to register. Required when enabled is true. Each distinct pipeline (application) consuming a stream must use its own consumer name, as Kinesis permits only one active subscription per consumer per shard. Instances of the same pipeline sharing a DynamoDB checkpoint table should share this name.

Type: string
Default: ""

enhanced_fan_out.consumer_activation_timeout​

The maximum amount of time to wait on connect for the registered consumer to become active before failing. Newly registered consumers on streams with many shards can take tens of seconds to activate.

Type: string
Default: "1m"

enhanced_fan_out.max_resubscribe_interval​

The ceiling on the exponential backoff between SubscribeToShard attempts after a subscription ends without delivering any events. This bounds how long a shard may sit unsubscribed after repeated failures.

Type: string
Default: "30s"

auto_replay_nacks​

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation.

Type: bool
Default: true

commit_period​

The period of time between each update to the checkpoint table.

Type: string
Default: "5s"

steal_grace_period​

Determines how long beyond the next commit period a client will wait when stealing a shard for the current owner to store a checkpoint. A longer value increases the time taken to balance shards but reduces the likelihood of processing duplicate messages.

Type: string
Default: "2s"

rebalance_period​

The period of time between each attempt to rebalance shards across clients.

Type: string
Default: "30s"

lease_period​

The period of time after which a client that has failed to update a shard checkpoint is assumed to be inactive.

Type: string
Default: "30s"

start_from_oldest​

Whether to consume from the oldest message when a sequence does not yet exist for the stream.

Type: bool
Default: true

region​

The AWS region to target.

Type: string

endpoint​

Allows you to specify a custom endpoint for the AWS API.

Type: string

tcp​

TCP socket configuration.

Type: object

tcp.connect_timeout​

Maximum amount of time a dial will wait for a connect to complete. Zero disables.

Type: string
Default: "0s"

tcp.keep_alive​

TCP keep-alive probe configuration.

Type: object

tcp.keep_alive.idle​

Duration the connection must be idle before sending the first keep-alive probe. Zero defaults to 15s. Negative values disable keep-alive probes.

Type: string
Default: "15s"

tcp.keep_alive.interval​

Duration between keep-alive probes. Zero defaults to 15s.

Type: string
Default: "15s"

tcp.keep_alive.count​

Maximum unanswered keep-alive probes before dropping the connection. Zero defaults to 9.

Type: int
Default: 9

tcp.tcp_user_timeout​

Maximum time to wait for acknowledgment of transmitted data before killing the connection. Linux-only (kernel 2.6.37+), ignored on other platforms. When enabled, keep_alive.idle must be greater than this value per RFC 5482. Zero disables.

Type: string
Default: "0s"

credentials​

Optional manual configuration of AWS credentials to use. More information can be found in this document.

Type: object

credentials.profile​

A profile from ~/.aws/credentials to use.

Type: string

credentials.id​

The ID of credentials to use.

Type: string

credentials.secret​

The secret for the credentials being used.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string

credentials.token​

The token for the credentials being used, required when using short term credentials.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string

credentials.from_ec2_role​

Use the credentials of a host EC2 machine configured to assume an IAM role associated with the instance.

Type: bool

credentials.role​

A role ARN to assume.

Type: string

credentials.role_external_id​

An external ID to provide when assuming a role.

Type: string

batching​

Allows you to configure a batching policy.

Type: object

batching.count​

A number of messages at which the batch should be flushed. If 0 disables count based batching.

Type: int
Default: 0

batching.byte_size​

An amount of bytes at which the batch should be flushed. If 0 disables size based batching.

Type: int
Default: 0

batching.period​

A period in which an incomplete batch should be flushed regardless of its size.

Type: string
Default: ""

batching.check​

A Bloblang query that should return a boolean value indicating whether a message should end a batch.

Type: string
Default: ""

batching.processors​

A list of processors to apply to a batch as it is flushed. This allows you to aggregate and archive the batch however you see fit. Please note that all resulting messages are flushed as a single batch, therefore splitting the batch into smaller batches using these processors is a no-op.

Type: array of processor