aws_kinesis: add enhanced fan-out support and poll_period

[squiidz]

Ticket: CON-541

Kinesis caps GetRecords at 5 calls/sec/shard, shared across every consumer of a stream, and the aws_kinesis input polls as fast as the pipeline drains with no way to slow it down — so a single Connect instance can starve co-tenant consumers, and adding more consumers causes ReadProvisionedThroughputExceeded throttling. This PR adds both remedies:

• poll_period (default 0s): a minimum interval between GetRecords calls per shard, for staying under the shared read budget. Validated against lease_period (rejected above it) and warned against commit_period; the gate is serviced in bounded slices so checkpointing, lease renewal, and batch delivery are never starved.
• enhanced_fan_out (default off): consumes streams via a dedicated EFO consumer (SubscribeToShard HTTP/2 push, 2 MB/s/shard dedicated throughput). The named consumer is auto-registered per stream (in parallel) and never deregistered. Subscriptions resubscribe at the continuation sequence with a 1s floor between attempts; shard-handoff contention (ResourceInUseException) is absorbed by background retries; sequences rejected by Kinesis fall back to the oldest retained record; latest-mode starts are anchored to a fixed timestamp so retries cannot skip records.

The consumer loop's record-fetch step is extracted behind a shardRecordSource interface. Polling behavior is preserved exactly, with one deliberate exception (also a pre-existing bug on main): a stored sequence that has aged out of the retention window now falls back to the oldest retained record instead of stalling the shard forever. Checkpointing, shard balancing, and ack semantics are unchanged; both features default off.

Testing: full unit suite green under -race; LocalStack integration — the pre-existing aws_kinesis suite passes unchanged (regression gate), plus new EFO (balanced + explicit-shard checkpoint capture) and poll_period suites. task fmt / task lint / task docs clean. Note: EFO is verified against LocalStack's kinesis-mock; a real-AWS smoke test covering a two-instance shard steal is recommended before GA, since kinesis-mock doesn't enforce the 5s subscription-takeover window.

[claude]

fetchTimeout is derived only from commit_period, but a blocking Fetch also delays the timed batch flush, not just checkpointing.

In runConsumer the pull happens at the top of the loop body, before the select that services nextTimedBatchChan:

connect/internal/impl/aws/kinesis/input.go

Lines 559 to 570 in 2910f76

	for {
	var err error
	if state == awsKinesisConsumerConsuming && len(pending) == 0 && nextPullChan == unblockedChan && pendingMsg.msg == nil {
	var done bool
	if pending, done, err = source.Fetch(k.ctx); err != nil {
	if !awsErrIsTimeout(err) {
	nextPullChan = time.After(boff.NextBackOff())
	k.log.Errorf("Failed to pull Kinesis records: %v\n", err)
	}
	} else if len(pending) == 0 {
	// A blocking source waits for data internally, so an empty
	// result must be retried immediately rather than backed off.

So once pending is drained and nextPullChan is unblocked, the loop calls source.Fetch, which for the EFO source blocks up to fetchTimeout (1s by default) and for the gated polling source sleeps up to maxGateWait. An already-expired nextTimedBatchChan is only observed after that returns, so a partial batch configured with e.g. batching.period: 100ms is flushed up to ~1s late whenever the shard is idle. Before this change getRecords was a non-long-polling call that returned in milliseconds, so this is a new latency regression on both new code paths.

Suggested fix: bound fetchTimeout by the configured batching period as well as commit_period/2 (or have the loop re-check nextTimedBatchChan before entering a blocking fetch).

[squiidz]

Good catch — fixed in c30536f. The fetch wait is now bounded by shardFetchWaitBound = min(1s, commit_period/2, batching.period/2) with a 5ms floor, applied to both the EFO fetch park and the poll_period gate, so timed batch flushes are serviced within half the batching period.

[claude]

These are user-facing timing knobs (how long Connect waits for a newly registered consumer to reach ACTIVE, and how often it polls DescribeStreamConsumer) but they're hardcoded package vars that are only mutable from tests.

The project Go patterns state under Configurable Time Parameters: "Every time-related value (timeouts, backoffs, intervals, retry delays) must be exposed as a YAML-configurable field. Do not hardcode durations." (.claude/agents/godev.md)

efoResubscribeFloor is arguably fine since it encodes a fixed AWS API limit, but efoConsumerActiveTimeout (and the boff.MaxInterval = time.Second * 30 in run) are policy choices a user may need to tune — a 1-minute cap on consumer activation will fail Connect outright on a slow registration. Consider exposing them under the enhanced_fan_out object.

[squiidz]

Addressed across c30536f and d45bdff: enhanced_fan_out.consumer_activation_timeout (default 1m) and enhanced_fan_out.max_resubscribe_interval (default 30s, validated >= the 1s floor) are now advanced fields, and the fetch park is no longer a literal — it's derived as min(1s, commit_period/2, batching.period/2). Deliberately kept internal: the 1s resubscribe floor (encodes the SubscribeToShard 1-call/sec/shard/consumer API limit — commented as such) and the DescribeStreamConsumer poll cadence (an implementation detail inside the configurable activation timeout, also commented). Remaining literals are named constants.

[claude]

The new enhanced fan-out path introduces several hardcoded durations that are only overridable from tests, not from YAML:

• efoConsumerActiveTimeout = time.Minute — this one is user-visible: it bounds how long Connect blocks per stream waiting for a newly registered consumer to reach ACTIVE, and on a slow/throttled account it decides whether startup fails or succeeds.
• efoConsumerPollInterval = time.Second — the DescribeStreamConsumer poll cadence during that wait.
• boff.MaxInterval = time.Second * 30 in run (input_efo.go#L717-L725) — the resubscribe backoff ceiling.
• fetchTimeout := time.Second in newShardRecordSource (input.go#L676-L689) — how long a single Fetch parks waiting for a pushed batch.

This conflicts with the project Go patterns in .claude/agents/godev.md, Configurable Time Parameters:

Every time-related value (timeouts, backoffs, intervals, retry delays) must be exposed as a YAML-configurable field. Do not hardcode durations.

Suggested fix: expose at least the consumer-activation timeout (and ideally the poll interval and resubscribe backoff ceiling) as fields under the enhanced_fan_out object, with the current values as static defaults, and derive fetchTimeout from a configurable field rather than a literal. efoResubscribeFloor is reasonable to leave fixed since it mirrors the AWS one-call-per-second-per-shard API limit — worth a short comment saying so.

[squiidz]

Addressed across c30536f and d45bdff: enhanced_fan_out.consumer_activation_timeout (default 1m) and enhanced_fan_out.max_resubscribe_interval (default 30s, validated >= the 1s floor) are now advanced fields, and the fetch park is no longer a literal — it's derived as min(1s, commit_period/2, batching.period/2). Deliberately kept internal: the 1s resubscribe floor (encodes the SubscribeToShard 1-call/sec/shard/consumer API limit — commented as such) and the DescribeStreamConsumer poll cadence (an implementation detail inside the configurable activation timeout, also commented). Remaining literals are named constants.

[claude]

Resubscribe backoff is never advanced when a subscription ends without an error and without progress

boff.NextBackOff() is only consumed inside this if streamErr != nil branch, but the reset is gated on sawEvent (input_efo.go#L317-L325). The two conditions are independent, so the case the comment explicitly calls out is not actually covered:

• consume returns because sub.Events() was closed, sawEvent == false (no SubscribeToShardEvent was ever delivered) and sub.Err() == nil.
• boff is neither reset nor advanced; the loop falls through to the top with sub == nil.
• waitForResubscribeFloor waits only efoResubscribeFloor (1s), and the cycle repeats indefinitely at exactly the one-call-per-second-per-shard SubscribeToShard limit.

The comment above states the intent as "a subscription that is accepted and then dies without delivering a single event would otherwise retry at the one-second floor forever, saturating the SubscribeToShard limit. Escalating instead also damps the subscription ping-pong two consumers can produce whilst a shard lease is being stolen." — the shard-steal ping-pong is precisely a case where the losing subscription can be terminated without a stream error, so it keeps hammering the floor rather than escalating.

Suggested fix: drive the escalation off progress rather than off the error, e.g. sleep boff.NextBackOff() whenever !sawEvent (and still on streamErr != nil), so a subscription that is accepted and dies with no records backs off as documented.

Relates to CONTRIBUTING.md §3.2.5 (excessive resource usage) and §3.1.4 (complete and correct implementation).

[squiidz]

Right — the errorless zero-event end wasn't covered. Fixed in 7bc168d: the backoff wait now applies whenever a subscription ends without having delivered an event (streamErr != nil || !sawEvent), so accepted-then-dead subscriptions escalate toward the ceiling as documented, which also damps the shard-steal ping-pong. Test added asserting escalation under repeated instant errorless closes.

[claude]

No benchmarks for the new enhanced fan-out consumption path (CONTRIBUTING.md §1.3.4/§1.3.5)

aws_kinesis (input) is classified certified in internal/plugins/info.csv, and §1.3.4 requires "two distinct phases, both of which are required" — local benchmarking and real-endpoint benchmarking — with results recorded under docs/benchmark-results/, plus §1.3.5 "benchmarks have been run at various throughput levels so that we can determine CPU and memory trendlines based on usage."

This PR introduces an entirely new consumption path (efoRecordSource, HTTP/2 push via SubscribeToShard) whose stated purpose is throughput — 2MB/s per shard vs. the shared 5 reads/sec/shard budget — plus a poll_period throttle, and no benchmark run or result file accompanies it. There is no kinesis entry under docs/benchmark-results/, so the throughput/CPU/memory claims in the docs and CHANGELOG are unverified.

Please follow the process in docs/benchmarking.md and record the EFO vs. polling comparison under docs/benchmark-results/.

[squiidz]

Fair point on §1.3.4–§1.3.6 — the local and real-endpoint benchmark phases for the EFO path (vs the polling path, poll_period overhead, and the KCL EFO consumer comparison) will be recorded under docs/benchmark-results/ as a follow-up before this ships in a release. The real-endpoint phase needs actual Kinesis: LocalStack's kinesis-mock can't demonstrate the dedicated-throughput claims, the genuine 5-minute subscription lifetime, or the 5s subscription-takeover window, so that run will double as the real-AWS validation of the resubscribe/steal behaviour.

[claude]

A capped gate wait returns nil, false, nil — indistinguishable from "the shard had no records". The consumer loop treats that as an empty poll and, because Blocking() is false for this source, arms its failure backoff:

connect/internal/impl/aws/kinesis/input.go

Lines 591 to 596 in 7bc168d

	} else if len(pending) == 0 {
	// A blocking source waits for data internally, so an empty
	// result must be retried immediately rather than backed off.
	if !source.Blocking() {
	nextPullChan = time.After(boff.NextBackOff())
	}

That backoff is the reader's shared boffPool entry (InitialInterval 300ms, MaxInterval 5s —

connect/internal/impl/aws/kinesis/input.go

Lines 385 to 393 in 7bc168d

	k.boffPool = sync.Pool{
	New: func() any {
	boff := backoff.NewExponentialBackOff()
	boff.InitialInterval = time.Millisecond * 300
	boff.MaxInterval = time.Second * 5
	boff.MaxElapsedTime = 0
	return boff
	},
	}

), and it is only Reset() when a fetch actually returns records. So whenever poll_period exceeds maxGateWait (which is 1s for the default commit_period: 5s, so any poll_period above 1s — including the poll_period: 10s case exercised in input_test.go), every gate-capped return escalates the backoff purely as a side effect of the user's configured period.

Concrete effect with poll_period: 10s, commit_period: 5s (so maxGateWait = 1s) on a shard that is producing: the first Fetch returns records and resets the backoff; subsequent fetches are capped at 1s and return empty, so the loop waits 300ms → 450ms → 675ms → … → 5s between attempts. The gate can only open on an attempt that lands after it elapses, so the effective spacing between GetRecords calls becomes roughly poll_period plus up to MaxInterval (~15s observed for a configured 10s), and records that arrive just after the gate opens sit undelivered for that extra interval. It also means an error-recovery backoff is escalating during entirely healthy operation.

Suggested fix: distinguish "gate not yet open" from "shard empty" — e.g. have Fetch signal the capped case separately (or return a sentinel the loop recognises) so the consumer loop re-enters Fetch promptly without arming boff, leaving the gate itself as the only pacing mechanism.

[squiidz]

Fixed in d45bdff: a gate-capped Fetch now returns a dedicated errPollGateWaiting sentinel which the consumer loop treats like a timeout — no failure backoff, no log — so the gate is the only pacing mechanism and GetRecords spacing tracks poll_period. A test asserts spacing stays within 1.5x the configured period under a tight retry loop.

[claude]

This warning tells operators something the implementation specifically prevents. shardFetchWaitBound caps a single Fetch wait at min(1s, commit_period/2, batch_period/2) and the poll gate returns early once that cap is hit:

connect/internal/impl/aws/kinesis/input_record_source.go

Lines 156 to 171 in 7bc168d

	if p.pollPeriod > 0 {
	if wait := p.pollPeriod - time.Since(p.lastPoll); wait > 0 {
	sleep, capped := wait, false
	if p.maxGateWait > 0 && p.maxGateWait < wait {
	sleep, capped = p.maxGateWait, true
	}
	select {
	case <-time.After(sleep):
	case <-ctx.Done():
	return nil, false, ctx.Err()
	}
	if capped {
	return nil, false, nil
	}
	}
	}

so control always returns to the consumer loop's select — which includes <-commitCtx.Done() — within half a commit period, regardless of how large poll_period is. Checkpoint updates and lease renewal are therefore not delayed by poll_period > commit_period; whatever bounded delay exists is commit_period/2 and is independent of poll_period. The same claim is repeated in the field description ("Values above commit_period delay checkpointing and shard-lease renewal") at

connect/internal/impl/aws/kinesis/input.go

Lines 199 to 204 in 7bc168d

	Description("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.").
	ShortDescription("Maximum gap between the in-flight sequence and the latest acknowledged sequence.").
	Default(1024),
	service.NewDurationField(kiFieldPollPeriod).
	Description("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. Values above `commit_period` delay checkpointing and shard-lease renewal, and values above `lease_period` are rejected.").
	ShortDescription("Minimum period between record polls of a shard, for staying under the shared Kinesis read limit.").

and in the generated docs.

This trips both CONTRIBUTING §1.2.2 ("Unexpected behavior should emit warning or error logs. Normal operation should emit no logs") — a valid, working config emits a warning on every startup — and §1.1.1/§1.1.2 for the documentation accuracy.

Suggested fix: either drop the warning and correct the field description, or reword both to describe the actual consequence (poll_period above commit_period means a shard may go a full poll period without new records, so the committed sequence advances no faster than that).

[squiidz]

Correct — that warning predated the fetch-wait cap and became false once shardFetchWaitBound landed. Fixed in d45bdff: the warning is removed and the field description now states the actual consequence (a shard is polled at most once per period, so the committed sequence advances no faster than that; values above lease_period are rejected).

[claude]

time.Second * 30 is a hardcoded, unnamed retry cap. The project Go patterns are explicit on both counts:

Configurable Time Parameters — Every time-related value (timeouts, backoffs, intervals, retry delays) must be exposed as a YAML-configurable field. Do not hardcode durations.

Magic Numbers — Name all numeric constants. Every literal number in logic must have a clear meaning through a named constant or variable.

This is the ceiling on how long a shard can sit unsubscribed after a failure, so an operator has no way to tighten or loosen recovery latency. Same applies to efoConsumerPollInterval = time.Second (

connect/internal/impl/aws/kinesis/input_efo.go

Lines 34 to 37 in 7bc168d

	var (
	// efoConsumerPollInterval is the internal poll cadence used while waiting
	// for a registered consumer to become active.
	efoConsumerPollInterval = time.Second

) and the bare time.Second default in shardFetchWaitBound (

connect/internal/impl/aws/kinesis/input.go

Lines 701 to 704 in 7bc168d

	// and timed batch flushes.
	func shardFetchWaitBound(commitPeriod, batchPeriod time.Duration) time.Duration {
	bound := time.Second
	if half := commitPeriod / 2; half < bound {

), where the companion minBound is named.

efoResubscribeFloor is fine as-is — it encodes an AWS API limit rather than a tuning choice, and the comment says so.

Suggested fix: name these as package constants at minimum, and expose the resubscribe backoff ceiling as an advanced field under enhanced_fan_out (alongside the consumer_activation_timeout this PR already added).

[squiidz]

Done in d45bdff: max_resubscribe_interval is exposed as an advanced field under enhanced_fan_out (default 30s, validated >= the 1s API floor so it can't undercut it), the bare time.Second in shardFetchWaitBound is now the named constant maxShardFetchWait, and efoConsumerPollInterval carries a comment explaining why it stays internal (it's the describe cadence inside the configurable consumer_activation_timeout).

[claude]

aws_kinesis (input) is marked certified in internal/plugins/info.csv:

connect/internal/plugins/info.csv

Lines 19 to 21 in 7bc168d

	aws_dynamodb_partiql ,processor ,aws_dynamodb_partiql ,certified ,n ,y ,y ,
	aws_kinesis ,input ,AWS Kinesis ,certified ,n ,y ,y ,
	aws_kinesis ,output ,AWS Kinesis ,certified ,n ,y ,y ,

The entire justification for this feature is throughput — the description and CHANGELOG both sell it as "2MB/s per shard … avoiding the shared 5 reads per second per shard limit" — yet the PR carries no benchmark evidence that the EFO path actually beats the polling path here. CONTRIBUTING §1.3.4 requires both phases (local benchmarking and real-endpoint benchmarking), §1.3.5 requires runs at various throughput levels for CPU/memory trendlines, and §1.3.6 requires a comparison against the equivalent Kafka Connect connector (the KCL EFO consumer) where one exists. Results are to be recorded under docs/benchmark-results/ — there is currently no Kinesis entry there.

Note rather than a blocker on the code itself, but for a perf-motivated change on a certified connector it is the evidence a reviewer needs: a real-endpoint run is also the only way to shake out the resubscribe/backoff behaviour under a genuine 5-minute AWS subscription lifetime, which LocalStack won't reproduce.

[squiidz]

Fair point on §1.3.4–§1.3.6 — the local and real-endpoint benchmark phases for the EFO path (vs the polling path, poll_period overhead, and the KCL EFO consumer comparison) will be recorded under docs/benchmark-results/ as a follow-up before this ships in a release. The real-endpoint phase needs actual Kinesis: LocalStack's kinesis-mock can't demonstrate the dedicated-throughput claims, the genuine 5-minute subscription lifetime, or the 5s subscription-takeover window, so that run will double as the real-AWS validation of the resubscribe/steal behaviour.

[claude]

Missing benchmarks for a throughput feature (CONTRIBUTING.md §1.3.4–§1.3.6)

aws_kinesis is a certified input (internal/plugins/info.csv:20), and this PR's entire purpose is read throughput — the CHANGELOG entry claims "2MB/s per shard of read throughput, avoiding the shared 5 reads per second per shard limit". CONTRIBUTING.md §1.3.4 requires two phases, both of which are required (local/Docker benchmarks plus real-endpoint benchmarks), §1.3.5 requires runs "at various throughput levels so that we can determine CPU and memory trendlines", and §1.3.6 requires a comparison against the equivalent Kafka Connect connector (which also supports enhanced fan-out).

There is no internal/impl/aws/kinesis/bench/ directory in this PR (the layout mandated by docs/benchmarking.md, and followed by e.g. internal/impl/aws/dynamodb/bench/ and internal/impl/aws/s3/bench/), and no docs/benchmark-results/aws-kinesis.md alongside the existing per-connector result files.

Suggested fix: add a bench/ suite for the Kinesis input covering both the polling path (with and without poll_period) and the enhanced fan-out path, run it at several throughput levels locally and against a real Kinesis stream, and record the results under docs/benchmark-results/ per §1.3.4. Without them there is no evidence that the new push path actually delivers the throughput the docs and CHANGELOG advertise, nor what its CPU/memory cost is versus polling.

[squiidz]

Fair point on §1.3.4–§1.3.6 — the local and real-endpoint benchmark phases for the EFO path (vs the polling path, poll_period overhead, and the KCL EFO consumer comparison) will be recorded under docs/benchmark-results/ as a follow-up before this ships in a release. The real-endpoint phase needs actual Kinesis: LocalStack's kinesis-mock can't demonstrate the dedicated-throughput claims, the genuine 5-minute subscription lifetime, or the 5s subscription-takeover window, so that run will double as the real-AWS validation of the resubscribe/steal behaviour.