Add aws_kinesis_streams source with KCL-style closed-shard completion.

[benmali]

Summary

This PR adds a new aws_kinesis_streams source. Vector can now read logs from AWS Kinesis
Data Streams.
The source stores progress in DynamoDB. Several Vector processes can share the same stream
and split shards between them.
Closed shards follow KCL 3.0 rules:

• A parent shard is finished with a SHARD_END checkpoint. Vector does not keep polling
  it until data expires.
• A child shard is not taken until the parent is SHARD_END (or has no lease).
• A child shard starts at TRIM_HORIZON, not LATEST.
• An empty GetRecords result on a closed shard stops that consumer.
• Old DynamoDB rows for shards that ListShards no longer returns are deleted.
  This stops CloudWatch GetRecords.IteratorAgeMilliseconds from growing for a long time
  after a reshard.
  Kinesis and DynamoDB errors now use the AWS SDK DisplayErrorContext helper. Logs show
  the full error chain, not only a short label like service error.
  Changelog files:
• changelog.d/aws_kinesis_streams_closed_shard_iterator.fix.md
• changelog.d/aws_kinesis_streams_sdk_error_context.enhancement.md

References

Vector configuration

sources:
  kinesis:
    type: aws_kinesis_streams
    region: us-east-1
    streams:
      - my-stream
    dynamodb:
      table: vector-kinesis-checkpoints
      create: true
    start_from_oldest: true
    checkpoint_limit: 1024
    max_records_per_call: 1000
    poll_interval_ms: 1000
    acknowledgements:
      enabled: true

sinks:
  console:
    type: console
    inputs:
      • kinesis

    encoding:
      codec: json

## How did you test this PR?
- Unit tests in `src/sources/aws_kinesis_streams/closed_shard.rs`. They cover closed vs
open shards, iterator choice (`TRIM_HORIZON` / `LATEST` / after sequence),
parent-before-child leases, `SHARD_END` cleanup, leftover DynamoDB rows, and when a
closed-shard consumer should stop.
- Changelog fragments added for the new source behavior and the clearer AWS error logs.
- Local setup: Rust Vector build with feature `sources-aws_kinesis_streams`.
I did not run AWS integration tests in this change. The existing
`aws-kinesis-streams-integration-tests` feature is still for the sink only.
## Does this PR include user facing changes?
- [x] Yes. Please add a changelog fragment based on our
[guidelines](https://github.com/vectordotdev/vector/blob/master/changelog.d/README.md).
- [ ] No. A maintainer will apply the `no-changelog` label to this PR.
This is user facing because it adds a new source and new config fields. AWS error logs
also show more detail.
## Contributor Guidelines
- Please read our [Vector contributor
resources](https://github.com/vectordotdev/vector/tree/master/docs#getting-started).
- Do not hesitate to use `@vectordotdev/vector` to reach out to us regarding this PR.
- Before pushing, follow our [pre-push
guidance](https://github.com/vectordotdev/vector/blob/master/CONTRIBUTING.md#pre-push).
- After a review is requested, please avoid force pushes to help us review incrementally.
  - Feel free to push as many commits as you want. They will be squashed into one before
merging.
  - For example, you can run `git merge origin master` and `git push`.

[chatgpt-codex-connector]

💡 Codex Review

vector/src/sources/aws_kinesis_streams/closed_shard.rs

Lines 165 to 168 in 55eb597

	if shard_closed && records_empty {
	return true;
	}
	growing_empty_polls >= GROWING_AGE_POLLS_TO_FINISH

Wait for the terminal iterator before ending a shard

An empty GetRecords response is not proof that a shard has been exhausted, even for a shard reported as closed, and increasing iterator age does not prove that an open shard was closed. Either branch can therefore set shard_finished, after which the cleanup writes SHARD_END; for the open-shard case this permanently stops the shard while it can still receive records, and for the closed-shard case it can skip records available after the empty page. Completion should require the authoritative missing next iterator rather than these empty-response heuristics.

---

vector/src/sources/aws_kinesis_streams/source.rs

Lines 1154 to 1156 in 55eb597

	let _ = checkpointer
	.checkpoint(&stream.id, &shard_id, SHARD_END, true)
	.await;

Drain final acknowledgements before writing SHARD_END

With acknowledgements enabled, the final GetRecords batch can still have an outstanding notifier when the response has no next iterator: the handler sends the batch, immediately leaves the loop, and unconditionally writes SHARD_END here. If downstream later rejects that batch or Vector exits before it is acknowledged, the checkpoint remains terminal and those records are never replayed. Wait for all in-flight records to resolve successfully before committing SHARD_END.

---

vector/src/sources/aws_kinesis_streams/batcher.rs

Lines 48 to 53 in 55eb597

	pub fn acknowledge(&self, count: i64, sequence: String) {
	self.in_flight.fetch_sub(count, Ordering::AcqRel);
	let mut acked = self.acked_sequence.lock().expect("tracker lock poisoned");
	if sequence > *acked {
	*acked = sequence;
	}

Advance checkpoints only across contiguous acknowledgements

When multiple batches are in flight, their downstream acknowledgements can complete out of order, but this immediately advances the checkpoint to whichever acknowledged sequence compares greatest. If a later batch is delivered while an earlier batch is still pending or rejected, the periodic DynamoDB checkpoint moves past the gap; a restart then begins after the later sequence and loses the earlier records. The tracker needs to retain batch order and advance only after every preceding batch has been delivered.

---

vector/src/sources/aws_kinesis_streams/source.rs

Lines 883 to 890 in 55eb597

	// Back-pressure: wait until there is room for a full batch before
	// issuing the next GetRecords call. Checking for max_records_per_call
	// rather than 1 prevents the in-flight count from significantly
	// exceeding checkpoint_limit when large batches are returned.
	if !tracker.can_accept(self.config.max_records_per_call as i64) {
	select! {
	_ = sleep(Duration::from_millis(10)) => { continue; }
	_ = cancel.cancelled() => { break; }

Bound each read by the remaining checkpoint capacity

Any documented checkpoint_limit smaller than max_records_per_call makes the source stall before its first request: at zero in-flight records, can_accept(max_records_per_call) is permanently false, so this loop only sleeps and never calls Kinesis. In particular, the advertised value checkpoint_limit: 1 cannot provide sequential processing with the default request limit of 1000. The request limit should be reduced to the remaining capacity rather than requiring capacity for the configured maximum.

---

vector/src/sources/aws_kinesis_streams/closed_shard.rs

Lines 165 to 168 in 55eb597

	if shard_closed && records_empty {
	return true;
	}
	growing_empty_polls >= GROWING_AGE_POLLS_TO_FINISH

Wait for the terminal iterator before ending a shard

An empty GetRecords response is not proof that a shard has been exhausted, even for a shard reported as closed, and increasing iterator age does not prove that an open shard was closed. Either branch can therefore set shard_finished, after which the cleanup writes SHARD_END; for the open-shard case this permanently stops the shard while it can still receive records, and for the closed-shard case it can skip records that a subsequent poll would have returned.

---

vector/src/sources/aws_kinesis_streams/source.rs

Lines 1154 to 1156 in 55eb597

	let _ = checkpointer
	.checkpoint(&stream.id, &shard_id, SHARD_END, true)
	.await;

Drain final acknowledgements before writing SHARD_END

When the final GetRecords response contains records and also indicates the end of the shard, its acknowledgement receiver is merely spawned before the loop breaks. Cleanup then writes SHARD_END immediately without waiting for those events to be delivered, so a crash or downstream rejection after this point permanently skips the final batch and unblocks child shards.

---

vector/src/sources/aws_kinesis_streams/batcher.rs

Lines 48 to 52 in 55eb597

	pub fn acknowledge(&self, count: i64, sequence: String) {
	self.in_flight.fetch_sub(count, Ordering::AcqRel);
	let mut acked = self.acked_sequence.lock().expect("tracker lock poisoned");
	if sequence > *acked {
	*acked = sequence;

Advance checkpoints only across contiguous acknowledgements

With acknowledgements enabled, receivers for multiple Kinesis batches run concurrently, so a later batch can be delivered before an earlier batch is delivered or rejected. This implementation immediately advances to the greatest reported sequence without tracking gaps; the next periodic checkpoint can therefore move past undelivered records, which are lost after a restart or lease transfer.

---

vector/src/sources/aws_kinesis_streams/source.rs

Lines 887 to 890 in 55eb597

	if !tracker.can_accept(self.config.max_records_per_call as i64) {
	select! {
	_ = sleep(Duration::from_millis(10)) => { continue; }
	_ = cancel.cancelled() => { break; }

Bound each read by the remaining checkpoint capacity

If checkpoint_limit is smaller than max_records_per_call—including the documented value checkpoint_limit: 1 intended for sequential processing—the initial in-flight count is zero but this condition is still always true. The consumer consequently sleeps forever without issuing its first GetRecords; the requested limit needs to be reduced to available capacity rather than requiring capacity for the configured maximum.

---

vector/src/sources/aws_kinesis_streams/closed_shard.rs

Lines 72 to 74 in 55eb597

	if shard_closed && !has_checkpoint {
	return false;
	}

Consume closed shards that lack checkpoints

On a fresh deployment or after recreating the checkpoint table, any shard that was closed by an earlier split or merge has no checkpoint and is discarded here. With the default start_from_oldest: true, those shards can still contain retained records that should be read; their children are simultaneously considered eligible because the missing parent checkpoint is treated as completed, so the historical parent records are permanently skipped.

---

vector/src/sources/aws_kinesis_streams/closed_shard.rs

Lines 183 to 189 in 55eb597

	pub fn iterator_error_action(shard_closed: bool, resource_not_found: bool) -> IteratorErrorAction {
	if resource_not_found {
	IteratorErrorAction::Delete
	} else if shard_closed {
	IteratorErrorAction::ShardEnd
	} else {
	IteratorErrorAction::Release

Retry transient iterator failures before ending closed shards

For a closed shard, every GetShardIterator failure other than ResourceNotFoundException is converted to ShardEnd, including transient transport failures, throttling, and temporary service errors. The caller then persists SHARD_END, so a single recoverable AWS error can permanently skip all records after the current checkpoint instead of being retried.

---

vector/src/sources/aws_kinesis_streams/source.rs

Lines 701 to 704 in 55eb597

	while !pending.is_empty() && !cancel.is_cancelled() {
	let mut still_pending = Vec::new();
	for (stream, shard_id) in pending.drain(..) {
	match checkpointer.claim(&stream.id, &shard_id, "").await {

Reclaim expired leases in explicit-shard mode

After an explicit-shard consumer crashes, its DynamoDB row retains ClientID; every restarted explicit consumer calls claim with an empty previous owner, whose condition requires attribute_not_exists(ClientID). Because this loop never examines LeaseTimeout or supplies the stale owner ID, it retries forever and the configured shard cannot be consumed again without manually editing the table.

---

vector/src/sources/aws_kinesis_streams/checkpointer.rs

Lines 512 to 517 in 55eb597

	self.svc
	.update_item()
	.table_name(&self.conf.table)
	.key("StreamID", AttributeValue::S(stream_id.to_string()))
	.key("ShardID", AttributeValue::S(shard_id.to_string()))
	.update_expression("SET SequenceNumber = :new_sequence_number")

Condition yield updates on the former ownership

When a consumer discovers that its conditional checkpoint failed because another client stole the lease, it calls this method against a row already owned by that new client. The unconditional update can overwrite the new owner's newer sequence—or even replace its SHARD_END marker—with the old consumer's sequence, causing replay after a crash and potentially blocking child-shard processing; the update must be conditional on an appropriate ownership/version state.

---

vector/src/sources/aws_kinesis_streams/closed_shard.rs

Lines 72 to 74 in 55eb597

	if shard_closed && !has_checkpoint {
	return false;
	}

Consume closed shards that lack checkpoints

When a source starts against an existing resharded stream with an empty or newly created checkpoint table, retained closed parent shards do not yet have checkpoint rows. Skipping them here loses all records still retained in those parents even with the default start_from_oldest: true, while their children are immediately considered eligible because the missing parent checkpoint is treated as completed.

---

vector/src/sources/aws_kinesis_streams/closed_shard.rs

Lines 183 to 189 in 55eb597

	pub fn iterator_error_action(shard_closed: bool, resource_not_found: bool) -> IteratorErrorAction {
	if resource_not_found {
	IteratorErrorAction::Delete
	} else if shard_closed {
	IteratorErrorAction::ShardEnd
	} else {
	IteratorErrorAction::Release

Retry transient iterator failures instead of ending shards

For a closed shard, every GetShardIterator failure other than ResourceNotFoundException is mapped to ShardEnd, including transient transport failures, throttling, and service errors. The caller then persists SHARD_END without reading the remaining records, so a temporary AWS failure can permanently discard retained data; only authoritative end-of-shard evidence should select this action.

---

vector/src/sources/aws_kinesis_streams/source.rs

Lines 701 to 705 in 55eb597

	while !pending.is_empty() && !cancel.is_cancelled() {
	let mut still_pending = Vec::new();
	for (stream, shard_id) in pending.drain(..) {
	match checkpointer.claim(&stream.id, &shard_id, "").await {
	Ok(seq) => {

Reclaim expired leases in explicit-shard mode

Explicit mode always calls claim with an empty previous-client ID, whose condition requires ClientID to be absent. If the previous Vector process crashes before releasing its lease, that attribute remains in DynamoDB forever; this retry loop never considers LeaseTimeout or supplies the stale owner ID, so the explicitly configured shard cannot be consumed again without manually editing the table.

---

vector/src/sources/aws_kinesis_streams/checkpointer.rs

Lines 512 to 517 in 55eb597

	self.svc
	.update_item()
	.table_name(&self.conf.table)
	.key("StreamID", AttributeValue::S(stream_id.to_string()))
	.key("ShardID", AttributeValue::S(shard_id.to_string()))
	.update_expression("SET SequenceNumber = :new_sequence_number")

Condition yielded checkpoints on the former ownership

When a consumer learns that another client has stolen its lease, yield_shard updates the row now owned by that new client without any ownership condition. If the new consumer has already checkpointed farther ahead, or has completed the shard with SHARD_END, this late write can regress or resurrect the checkpoint, causing replay and potentially blocking child shards; the update must be conditional on the expected ownership/state.

---

vector/src/sources/aws_kinesis_streams/source.rs

Lines 712 to 714 in 55eb597

	src.run_shard_consumer(
	stream, shard_id, seq, chk, out2, cancel2, false, false,
	)

Preserve shard topology in explicit mode

Every explicit shard is launched with shard_closed and is_child hard-coded to false instead of using its ListShards metadata. Consequently, with start_from_oldest: false, an explicitly selected closed shard or child requests a LATEST iterator and skips all records already retained there; explicit mode also bypasses the parent-completion rule that balanced mode applies.

---

vector/src/sources/aws_kinesis_streams/source.rs

Lines 1201 to 1204 in 55eb597

	if let Event::Log(ref mut log) = event {
	match self.log_namespace {
	LogNamespace::Vector => {
	if let Some(ts) = timestamp {

Insert the standard source metadata on decoded logs

The declared output schema includes standard Vector source metadata, but decoded logs only receive vector.ingest_timestamp; neither the Vector namespace nor legacy path receives source_type. Downstream transforms that route on the standard source type therefore see the field missing despite the schema promising it, so this path should insert the standard metadata in the same way as other sources.

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".