aws_kinesis: add enhanced fan-out support and poll_period

CHANGELOG.md

@@ -3,6 +3,16 @@ Changelog
 
 All notable changes to this project will be documented in this file.
 
+## Unreleased
+
+### Added
+
+- aws_kinesis: Added a `poll_period` field to bound the rate of `GetRecords` calls per shard, and an `enhanced_fan_out` configuration block that consumes streams via a dedicated enhanced fan-out consumer with 2MB/s per shard of read throughput, avoiding the shared 5 reads per second per shard limit. ([@squiidz](https://github.com/squiidz), [#4724](https://github.com/redpanda-data/connect/pull/4724))
+
+### Fixed
+
+- aws_kinesis: The input now falls back to the oldest retained record when a stored sequence has aged out of the stream's retention window, instead of retrying the stale position indefinitely. ([@squiidz](https://github.com/squiidz), [#4724](https://github.com/redpanda-data/connect/pull/4724))
+
 ## 4.106.0 - 2026-08-20
 
 ### Added

docs/modules/components/pages/inputs/aws_kinesis.adoc

@@ -90,6 +90,12 @@ input:
         role: "" # No default (optional)
         role_external_id: "" # No default (optional)
     checkpoint_limit: 1024
+    poll_period: 0s
+    enhanced_fan_out:
+      enabled: false
+      consumer_name: ""
+      consumer_activation_timeout: 1m
+      max_resubscribe_interval: 30s
     auto_replay_nacks: true
     commit_period: 5s
     steal_grace_period: 2s
@@ -132,6 +138,13 @@ Redpanda Connect will not store a consumed sequence unless it is acknowledged at
 
 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`, `kinesis:RegisterStreamConsumer` 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.
+
 == Table schema
 
 It's possible to configure Redpanda Connect 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`.
@@ -373,6 +386,61 @@ The maximum gap between the in flight sequence versus the latest acknowledged se
 
 *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"`
+Requires version 4.107.0 or newer
+
+=== `enhanced_fan_out`
+
+Consume the stream using https://docs.aws.amazon.com/streams/latest/dev/enhanced-consumers.html[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`, `kinesis:RegisterStreamConsumer` and `kinesis:SubscribeToShard`. Note that AWS bills enhanced fan-out consumers per consumer-shard-hour plus data retrieval.
+
+
+*Type*: `object`
+
+Requires version 4.107.0 or newer
+
+=== `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.