diff --git a/CHANGELOG.md b/CHANGELOG.md index 6bb7e3a8e3..e731986733 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/modules/components/pages/inputs/aws_kinesis.adoc b/docs/modules/components/pages/inputs/aws_kinesis.adoc index 2823ea4e67..fb2ebff683 100644 --- a/docs/modules/components/pages/inputs/aws_kinesis.adoc +++ b/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. diff --git a/internal/impl/aws/kinesis/input.go b/internal/impl/aws/kinesis/input.go index 2845c50ba0..d1de29b051 100644 --- a/internal/impl/aws/kinesis/input.go +++ b/internal/impl/aws/kinesis/input.go @@ -51,22 +51,35 @@ const ( kiFieldLeasePeriod = "lease_period" kiFieldRebalancePeriod = "rebalance_period" kiFieldStartFromOldest = "start_from_oldest" + kiFieldPollPeriod = "poll_period" + kiFieldEnhancedFanOut = "enhanced_fan_out" kiFieldBatching = "batching" + // Kinesis Enhanced Fan-Out Fields + kiefoFieldEnabled = "enabled" + kiefoFieldConsumerName = "consumer_name" + kiefoFieldActivationTimeout = "consumer_activation_timeout" + kiefoFieldMaxResubscribeInterval = "max_resubscribe_interval" + // Kinesis metrics metricShardsPerClient = "kinesis_client_shards" metricShardsStolen = "kinesis_shards_stolen_total" ) type kiConfig struct { - Streams []string - DynamoDB kiddbConfig - CheckpointLimit int - CommitPeriod string - StealGracePeriod string - LeasePeriod string - RebalancePeriod string - StartFromOldest bool + Streams []string + DynamoDB kiddbConfig + CheckpointLimit int + CommitPeriod string + StealGracePeriod string + LeasePeriod string + RebalancePeriod string + StartFromOldest bool + PollPeriod time.Duration + EFOEnabled bool + EFOConsumerName string + EFOActivationTimeout time.Duration + EFOMaxResubscribeInterval time.Duration } func kinesisInputConfigFromParsed(pConf *service.ParsedConfig) (conf kiConfig, err error) { @@ -96,6 +109,36 @@ func kinesisInputConfigFromParsed(pConf *service.ParsedConfig) (conf kiConfig, e if conf.StartFromOldest, err = pConf.FieldBool(kiFieldStartFromOldest); err != nil { return } + if conf.PollPeriod, err = pConf.FieldDuration(kiFieldPollPeriod); err != nil { + return + } + { + efoConf := pConf.Namespace(kiFieldEnhancedFanOut) + if conf.EFOEnabled, err = efoConf.FieldBool(kiefoFieldEnabled); err != nil { + return + } + if conf.EFOConsumerName, err = efoConf.FieldString(kiefoFieldConsumerName); err != nil { + return + } + if conf.EFOEnabled && conf.EFOConsumerName == "" { + err = fmt.Errorf("%v.%v is required when %v.%v is true", kiFieldEnhancedFanOut, kiefoFieldConsumerName, kiFieldEnhancedFanOut, kiefoFieldEnabled) + return + } + if conf.EFOActivationTimeout, err = efoConf.FieldDuration(kiefoFieldActivationTimeout); err != nil { + return + } + if conf.EFOEnabled && conf.EFOActivationTimeout <= 0 { + err = fmt.Errorf("%v.%v must be greater than zero", kiFieldEnhancedFanOut, kiefoFieldActivationTimeout) + return + } + if conf.EFOMaxResubscribeInterval, err = efoConf.FieldDuration(kiefoFieldMaxResubscribeInterval); err != nil { + return + } + if conf.EFOEnabled && conf.EFOMaxResubscribeInterval < time.Second { + err = fmt.Errorf("%v.%v must be at least 1s", kiFieldEnhancedFanOut, kiefoFieldMaxResubscribeInterval) + return + } + } return } @@ -114,6 +157,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`"+`. @@ -158,6 +208,34 @@ Use the `+"`batching`"+` fields to configure an optional xref:configuration:batc 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. A shard is polled at most once per period, so the committed sequence advances no faster than that; values above `lease_period` are rejected."). + ShortDescription("Minimum period between record polls of a shard, for staying under the shared Kinesis read limit."). + Default("0s"). + Version("4.107.0"). + Advanced(), + service.NewObjectField(kiFieldEnhancedFanOut, + service.NewBoolField(kiefoFieldEnabled). + Description("Whether to consume the stream using enhanced fan-out."). + Default(false), + service.NewStringField(kiefoFieldConsumerName). + Description("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."). + ShortDescription("The name of the enhanced fan-out consumer to register, unique per distinct pipeline."). + Default(""), + service.NewDurationField(kiefoFieldActivationTimeout). + Description("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."). + Default("1m"). + Advanced(), + service.NewDurationField(kiefoFieldMaxResubscribeInterval). + Description("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."). + Default("30s"). + Advanced(), + ). + Description("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."). + ShortDescription("Consume the stream using enhanced fan-out for dedicated read throughput."). + Version("4.107.0"). + Advanced(). + LintRule(`root = if this.enabled && this.consumer_name == "" { [ "consumer_name is required when enabled is true" ] }`), service.NewAutoRetryNacksToggleField(), service.NewDurationField(kiFieldCommitPeriod). Description("The period of time between each update to the checkpoint table."). @@ -207,6 +285,7 @@ type streamInfo struct { explicitShards []string id string // Either a name or arn, extracted from config and used for balancing shards arn string + consumerARN string // Enhanced fan-out consumer ARN, set when EFO is enabled } type kinesisReader struct { @@ -230,6 +309,8 @@ type kinesisReader struct { stealGracePeriod time.Duration leasePeriod time.Duration rebalancePeriod time.Duration + pollPeriod time.Duration + batchPeriod time.Duration cMut sync.Mutex msgChan chan asyncMessage @@ -373,6 +454,19 @@ func newKinesisReaderFromConfig(conf kiConfig, batcher service.BatchPolicy, sess if k.rebalancePeriod, err = time.ParseDuration(k.conf.RebalancePeriod); err != nil { return nil, fmt.Errorf("parsing rebalance period string: %v", err) } + // batcher.Period is an optional Go-duration string on service.BatchPolicy; + // an empty string means no timed flush is configured. A value that fails + // to parse is left at zero here: the batcher's own construction parses the + // same string and surfaces the error, and this field only feeds the + // fetch-wait bound below, so a conservative zero is harmless. + if batcher.Period != "" { + k.batchPeriod, _ = time.ParseDuration(batcher.Period) + } + + k.pollPeriod = conf.PollPeriod + if k.pollPeriod > k.leasePeriod { + return nil, fmt.Errorf("%v (%v) must not exceed %v (%v)", kiFieldPollPeriod, k.pollPeriod, kiFieldLeasePeriod, k.leasePeriod) + } // Initialize metrics k.clientShardsMetric = mgr.Metrics().NewGauge(metricShardsPerClient) @@ -389,76 +483,6 @@ const ( ErrCodeKMSThrottlingException = "KMSThrottlingException" ) -func (k *kinesisReader) getIter(info streamInfo, shardID, sequence string) (string, error) { - iterType := types.ShardIteratorTypeTrimHorizon - if !k.conf.StartFromOldest { - iterType = types.ShardIteratorTypeLatest - } - var startingSequence *string - if sequence != "" { - iterType = types.ShardIteratorTypeAfterSequenceNumber - startingSequence = &sequence - } - - res, err := k.svc.GetShardIterator(k.ctx, &kinesis.GetShardIteratorInput{ - StreamARN: &info.arn, - ShardId: &shardID, - StartingSequenceNumber: startingSequence, - ShardIteratorType: iterType, - }) - if err != nil { - return "", err - } - - var iter string - if res.ShardIterator != nil { - iter = *res.ShardIterator - } - if iter == "" { - // If we failed to obtain from a sequence we start from beginning - iterType = types.ShardIteratorTypeTrimHorizon - - res, err := k.svc.GetShardIterator(k.ctx, &kinesis.GetShardIteratorInput{ - StreamARN: &info.arn, - ShardId: &shardID, - ShardIteratorType: iterType, - }) - if err != nil { - return "", err - } - - if res.ShardIterator != nil { - iter = *res.ShardIterator - } - } - if iter == "" { - return "", errors.New("obtaining shard iterator") - } - return iter, nil -} - -// IMPORTANT TO NOTE: The returned shard iterator (second return parameter) will -// always be the input iterator when the error parameter is nil, therefore -// replacing the current iterator with this return param should always be safe. -// -// Do NOT modify this method without preserving this behaviour. -func (k *kinesisReader) getRecords(info streamInfo, shardIter string) ([]types.Record, string, error) { - res, err := k.svc.GetRecords(k.ctx, &kinesis.GetRecordsInput{ - StreamARN: &info.arn, - Limit: &awsKinesisDefaultLimit, - ShardIterator: &shardIter, - }) - if err != nil { - return nil, shardIter, err - } - - nextIter := "" - if res.NextShardIterator != nil { - nextIter = *res.NextShardIterator - } - return res.Records, nextIter, nil -} - func awsErrIsTimeout(err error) bool { return errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || @@ -496,8 +520,8 @@ func (k *kinesisReader) runConsumer(wg *sync.WaitGroup, info streamInfo, shardID // Stores consumed records that have yet to be added to the batcher. var pending []types.Record - var iter string - if iter, initErr = k.getIter(info, shardID, startingSequence); initErr != nil { + var source shardRecordSource + if source, initErr = k.newShardRecordSource(info, shardID, startingSequence, recordBatcher.GetSequence); initErr != nil { return initErr } @@ -527,6 +551,7 @@ func (k *kinesisReader) runConsumer(wg *sync.WaitGroup, info streamInfo, shardID defer func() { commitCtxClose() recordBatcher.Close(context.Background(), state == awsKinesisConsumerFinished) + source.Close() boff.Reset() k.boffPool.Put(boff) @@ -566,37 +591,34 @@ func (k *kinesisReader) runConsumer(wg *sync.WaitGroup, info streamInfo, shardID for { var err error - if state == awsKinesisConsumerConsuming && len(pending) == 0 && nextPullChan == unblockedChan { - if pending, iter, err = k.getRecords(info, iter); err != nil { - if !awsErrIsTimeout(err) { + 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) && !errors.Is(err, errPollGateWaiting) { nextPullChan = time.After(boff.NextBackOff()) - - var aerr *types.ExpiredIteratorException - if errors.As(err, &aerr) { - k.log.Warn("Shard iterator expired, attempting to refresh") - newIter, err := k.getIter(info, shardID, recordBatcher.GetSequence()) - if err != nil { - k.log.Errorf("Failed to refresh shard iterator: %v", err) - } else { - iter = newIter - } - } else { - k.log.Errorf("Failed to pull Kinesis records: %v\n", err) - } + k.log.Errorf("Failed to pull Kinesis records: %v\n", err) } } else if len(pending) == 0 { - nextPullChan = time.After(boff.NextBackOff()) + // 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()) + } } else { boff.Reset() nextPullChan = blockedChan } - // The getRecords method ensures that it returns the input - // iterator whenever it errors out. Therefore, regardless of the - // outcome of the call if iter is now empty we have definitely - // reached the end of the shard. - if iter == "" { + if done { state = awsKinesisConsumerFinished } + } else if pendingMsg.msg != nil { + // Park pulls while a message awaits delivery so that a + // blocking Fetch cannot delay its flush, and so the + // always-ready unblocked channel cannot outcompete the + // flush case in the select below. + if nextPullChan == unblockedChan { + nextPullChan = blockedChan + } } else { unblockPullChan() } @@ -684,6 +706,44 @@ func (k *kinesisReader) runConsumer(wg *sync.WaitGroup, info streamInfo, shardID return nil } +// maxShardFetchWait is the initial (commit/batch-period-independent) bound on +// how long a single Fetch may block, companion to the floor enforced by +// minBound below. +const maxShardFetchWait = time.Second + +// shardFetchWaitBound bounds how long a shard record source may internally +// block per Fetch so that the consumer loop keeps servicing its commit timer +// and timed batch flushes. +func shardFetchWaitBound(commitPeriod, batchPeriod time.Duration) time.Duration { + bound := maxShardFetchWait + if half := commitPeriod / 2; half < bound { + bound = half + } + if batchPeriod > 0 { + if half := batchPeriod / 2; half < bound { + bound = half + } + } + const minBound = 5 * time.Millisecond + if bound < minBound { + bound = minBound + } + return bound +} + +// newShardRecordSource creates the record source for a single claimed shard. +func (k *kinesisReader) newShardRecordSource(info streamInfo, shardID, startingSequence string, sequenceFn func() string) (shardRecordSource, error) { + // Bound how long a single Fetch may wait internally, whether that's the + // enhanced fan-out push wait or the poll_period gate, so the consumer loop + // keeps servicing its commit timer and timed batch flushes well within + // each period. + fetchTimeout := shardFetchWaitBound(k.commitPeriod, k.batchPeriod) + if k.conf.EFOEnabled { + return newEFORecordSource(k.ctx, kinesisEFOSubscribeFn(k.svc, info.consumerARN, shardID), shardID, startingSequence, k.conf.StartFromOldest, fetchTimeout, k.conf.EFOMaxResubscribeInterval, k.log) + } + return newPollingRecordSource(k.ctx, k.svc, info.arn, shardID, startingSequence, k.conf.StartFromOldest, k.pollPeriod, fetchTimeout, sequenceFn, k.log) +} + //------------------------------------------------------------------------------ func isShardFinished(s types.Shard) bool { @@ -947,12 +1007,50 @@ func (k *kinesisReader) Connect(ctx context.Context) error { k.svc = svc k.checkpointer = checkpointer - k.msgChan = make(chan asyncMessage) if err = k.waitUntilStreamsExists(ctx); err != nil { return err } + if k.conf.EFOEnabled { + // Registering a consumer involves waiting for it to become ACTIVE, so + // resolve every stream concurrently rather than serialising a minute + // long wait per stream. + results := make(chan error, len(k.streams)) + for _, s := range k.streams { + go func(info *streamInfo) { + // A previous Connect attempt may already have resolved this + // stream, in which case there is nothing left to do. + if info.consumerARN != "" { + results <- nil + return + } + arn, err := ensureEFOConsumer(ctx, svc, info.arn, k.conf.EFOConsumerName, k.conf.EFOActivationTimeout, k.log) + if err == nil { + info.consumerARN = arn + } + results <- err + }(s) + } + + // Every goroutine is drained before returning so that no writer to + // info.consumerARN outlives this call and races a retried Connect. + var efoErr error + for range k.streams { + if err := <-results; err != nil && efoErr == nil { + efoErr = err + } + } + if efoErr != nil { + return efoErr + } + } + + // Only mark the connection as established once every fallible step above + // has succeeded; otherwise a retried Connect would see a non-nil msgChan + // and return early without ever starting the shard runners. + k.msgChan = make(chan asyncMessage) + if len(k.streams[0].explicitShards) > 0 { go k.runExplicitShards() } else { diff --git a/internal/impl/aws/kinesis/input_efo.go b/internal/impl/aws/kinesis/input_efo.go new file mode 100644 index 0000000000..b787309dd0 --- /dev/null +++ b/internal/impl/aws/kinesis/input_efo.go @@ -0,0 +1,434 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package kinesis + +import ( + "context" + "errors" + "fmt" + "sync" + "sync/atomic" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/kinesis" + "github.com/aws/aws-sdk-go-v2/service/kinesis/types" + "github.com/cenkalti/backoff/v4" + + "github.com/redpanda-data/benthos/v4/public/service" +) + +// Overridable in tests. +var ( + // efoConsumerPollInterval is the internal poll cadence used while waiting + // for a registered consumer to become active. It stays internal (rather + // than a config field) because it is bounded by the user-configurable + // consumer_activation_timeout: it only affects how promptly activation is + // noticed within that window, not the overall time budget, so exposing it + // for independent tuning would add a knob with no real effect on outcomes. + efoConsumerPollInterval = time.Second + // efoResubscribeFloor is the minimum spacing enforced between successive + // SubscribeToShard calls for a given shard/consumer, matching the API's + // one-call-per-second limit. It also seeds the resubscribe backoff's + // initial interval. + efoResubscribeFloor = time.Second +) + +// efoConsumerAPI is the subset of the Kinesis API used to resolve and +// register enhanced fan-out stream consumers. +type efoConsumerAPI interface { + DescribeStreamConsumer(ctx context.Context, params *kinesis.DescribeStreamConsumerInput, optFns ...func(*kinesis.Options)) (*kinesis.DescribeStreamConsumerOutput, error) + RegisterStreamConsumer(ctx context.Context, params *kinesis.RegisterStreamConsumerInput, optFns ...func(*kinesis.Options)) (*kinesis.RegisterStreamConsumerOutput, error) +} + +// ensureEFOConsumer returns the ARN of the named enhanced fan-out consumer on +// the given stream, registering it if it does not exist and waiting for it to +// become ACTIVE. The consumer is intentionally never deregistered: multiple +// instances of the same pipeline share it as a single logical application. +func ensureEFOConsumer(ctx context.Context, api efoConsumerAPI, streamARN, name string, activationTimeout time.Duration, log *service.Logger) (string, error) { + ctx, cancel := context.WithTimeout(ctx, activationTimeout) + defer cancel() + + registered := false + for { + res, err := api.DescribeStreamConsumer(ctx, &kinesis.DescribeStreamConsumerInput{ + StreamARN: &streamARN, + ConsumerName: &name, + }) + if err != nil { + var nf *types.ResourceNotFoundException + if !errors.As(err, &nf) { + return "", fmt.Errorf("describing enhanced fan-out consumer '%v' on stream '%v' (requires kinesis:DescribeStreamConsumer): %w", name, streamARN, err) + } + if !registered { + if _, err := api.RegisterStreamConsumer(ctx, &kinesis.RegisterStreamConsumerInput{ + StreamARN: &streamARN, + ConsumerName: &name, + }); err != nil { + var inUse *types.ResourceInUseException + if !errors.As(err, &inUse) { + return "", fmt.Errorf("registering enhanced fan-out consumer '%v' on stream '%v' (requires kinesis:RegisterStreamConsumer): %w", name, streamARN, err) + } + // Another instance registered it concurrently; poll for it. + } else { + log.Infof("Registered Kinesis enhanced fan-out consumer '%v' on stream '%v'", name, streamARN) + } + registered = true + } + } else { + // Guard against quirky API responses with nil ConsumerDescription. + if res.ConsumerDescription == nil { + // Not ready yet; poll until it is. + } else { + switch res.ConsumerDescription.ConsumerStatus { + case types.ConsumerStatusActive: + if res.ConsumerDescription.ConsumerARN == nil { + return "", fmt.Errorf("enhanced fan-out consumer '%v' on stream '%v' returned ACTIVE status but nil ARN", name, streamARN) + } + return *res.ConsumerDescription.ConsumerARN, nil + case types.ConsumerStatusDeleting: + return "", fmt.Errorf("enhanced fan-out consumer '%v' on stream '%v' is currently being deleted, wait for the deletion to complete", name, streamARN) + } + // CREATING: poll until ACTIVE. + } + } + + select { + case <-time.After(efoConsumerPollInterval): + case <-ctx.Done(): + return "", fmt.Errorf("waiting for enhanced fan-out consumer '%v' on stream '%v' to become active: %w", name, streamARN, ctx.Err()) + } + } +} + +// efoSubscription is the consumable side of a SubscribeToShard event stream, +// satisfied by *kinesis.SubscribeToShardEventStream. +type efoSubscription interface { + Events() <-chan types.SubscribeToShardEventStream + Close() error + Err() error +} + +// efoSubscribeFn opens a shard subscription at the given position. +type efoSubscribeFn func(ctx context.Context, pos types.StartingPosition) (efoSubscription, error) + +// kinesisEFOSubscribeFn builds an efoSubscribeFn backed by the real Kinesis +// SubscribeToShard API for a registered consumer. +func kinesisEFOSubscribeFn(svc *kinesis.Client, consumerARN, shardID string) efoSubscribeFn { + return func(ctx context.Context, pos types.StartingPosition) (efoSubscription, error) { + out, err := svc.SubscribeToShard(ctx, &kinesis.SubscribeToShardInput{ + ConsumerARN: &consumerARN, + ShardId: &shardID, + StartingPosition: &pos, + }) + if err != nil { + return nil, err + } + return out.GetStream(), nil + } +} + +// efoRecordSource consumes a shard via enhanced fan-out push delivery. A +// background goroutine owns the subscription, forwarding record batches into +// a buffered channel and resubscribing at the continuation sequence whenever +// AWS terminates the subscription (roughly every five minutes). Backpressure +// from an unread channel simply pauses event consumption; flow control of +// in-flight messages remains governed by checkpoint_limit. +type efoRecordSource struct { + subscribe efoSubscribeFn + shardID string + fetchTimeout time.Duration + maxResubscribeInterval time.Duration + log *service.Logger + + recordsChan chan []types.Record + finished atomic.Bool + + ctx context.Context //nolint:containedctx // lifecycle context for the pump goroutine + cancel context.CancelFunc + wg sync.WaitGroup +} + +// efoStartingPosition computes the StartingPosition for a shard given its +// checkpointed sequence (if any) and the start_from_oldest setting. +// +// When there is no checkpoint and startFromOldest is false, the position is +// anchored to the current wall-clock time (AT_TIMESTAMP) rather than LATEST. +// LATEST is re-evaluated at subscribe time, so if the initial subscription +// dies before delivering any event, a naive retry using LATEST again would +// re-anchor at the new (later) tip and skip any records published in the +// interim. Anchoring a timestamp once, up front, is stable across retries: +// it means "latest as of construction" and never moves, so no records can be +// skipped between subscribe attempts. +func efoStartingPosition(startingSequence string, startFromOldest bool) types.StartingPosition { + if startingSequence != "" { + return types.StartingPosition{ + Type: types.ShardIteratorTypeAfterSequenceNumber, + SequenceNumber: &startingSequence, + } + } + if startFromOldest { + return types.StartingPosition{Type: types.ShardIteratorTypeTrimHorizon} + } + return types.StartingPosition{Type: types.ShardIteratorTypeAtTimestamp, Timestamp: aws.Time(time.Now())} +} + +// newEFORecordSource makes the first subscription attempt synchronously, so +// that ordinary misconfiguration (missing consumer, IAM) fails the shard +// claim fast, then starts the pump goroutine. +// +// Two kinds of failure on that first attempt are handled specially rather +// than failing the claim outright: +// +// - InvalidArgumentException while starting from a checkpointed sequence +// number: the stored sequence can no longer be resolved (e.g. it fell +// behind the shard's retention window). This mirrors the polling +// source's TRIM_HORIZON fallback: we log a warning and retry once from +// TRIM_HORIZON rather than failing the claim forever. The fallback is +// unconditionally TRIM_HORIZON (never the timestamp anchor above) because +// this shard demonstrably had a committed position: resuming at "now" +// would silently skip every record still retained ahead of the expired +// sequence, breaking at-least-once delivery. +// - ResourceInUseException: the shard's previous owner likely still holds +// the one allowed subscription during a steal/handoff. Rather than +// blocking the caller (and therefore the sequential claim/steal loop in +// runBalancedShards) waiting out AWS's cooldown, the pump is started +// with no live subscription; run's own resubscribe loop (spaced by +// efoResubscribeFloor, ctx-aware, indefinite) acquires it in the +// background. +// +// Any other error still fails fast. +func newEFORecordSource(ctx context.Context, subscribe efoSubscribeFn, shardID, startingSequence string, startFromOldest bool, fetchTimeout, maxResubscribeInterval time.Duration, log *service.Logger) (*efoRecordSource, error) { + pos := efoStartingPosition(startingSequence, startFromOldest) + + e := &efoRecordSource{ + subscribe: subscribe, + shardID: shardID, + fetchTimeout: fetchTimeout, + maxResubscribeInterval: maxResubscribeInterval, + log: log, + recordsChan: make(chan []types.Record, 1), + } + e.ctx, e.cancel = context.WithCancel(ctx) + + subscribedAt := time.Now() + sub, err := e.subscribe(e.ctx, pos) + + if err != nil && pos.Type == types.ShardIteratorTypeAfterSequenceNumber { + var invalidArg *types.InvalidArgumentException + if errors.As(err, &invalidArg) { + log.Warnf("Stored position for shard '%v' was rejected, falling back to the oldest retained record", shardID) + pos = types.StartingPosition{Type: types.ShardIteratorTypeTrimHorizon} + subscribedAt = time.Now() + sub, err = e.subscribe(e.ctx, pos) + } + } + + if err != nil { + var inUse *types.ResourceInUseException + if !errors.As(err, &inUse) { + e.cancel() + return nil, fmt.Errorf("subscribing to shard '%v' (requires kinesis:SubscribeToShard): %w", shardID, err) + } + sub = nil + } + + e.wg.Add(1) + go e.run(sub, pos, subscribedAt) + return e, nil +} + +// waitForResubscribeFloor blocks until at least efoResubscribeFloor has +// elapsed since lastSubscribeAt, so that SubscribeToShard is never called +// more than once per second for this shard/consumer regardless of how the +// previous subscription ended. Returns false if the context was cancelled +// first. +func (e *efoRecordSource) waitForResubscribeFloor(lastSubscribeAt time.Time) bool { + if wait := efoResubscribeFloor - time.Since(lastSubscribeAt); wait > 0 { + select { + case <-time.After(wait): + case <-e.ctx.Done(): + return false + } + } + return true +} + +func (e *efoRecordSource) run(sub efoSubscription, pos types.StartingPosition, lastSubscribeAt time.Time) { + defer func() { + close(e.recordsChan) + e.wg.Done() + }() + + boff := backoff.NewExponentialBackOff() + // SubscribeToShard is limited to one call per second per consumer per + // shard, so never retry faster than that. RandomizationFactor is zeroed + // so jitter never pulls a retry below that floor. + boff.InitialInterval = efoResubscribeFloor + boff.MaxInterval = e.maxResubscribeInterval + boff.MaxElapsedTime = 0 + boff.RandomizationFactor = 0 + boff.Reset() + + for { + if sub == nil { + if !e.waitForResubscribeFloor(lastSubscribeAt) { + return + } + lastSubscribeAt = time.Now() + var err error + if sub, err = e.subscribe(e.ctx, pos); err != nil { + if e.ctx.Err() != nil { + return + } + e.log.Errorf("Failed to subscribe to shard '%v': %v", e.shardID, err) + // A sequence-derived position that AWS refuses can never + // succeed on a retry (it has aged out of the shard's retention + // window), so fall back to the oldest retained record rather + // than wedging this shard forever whilst the lease renews. + // After the fallback the position is TRIM_HORIZON, so this can + // only fire once per stale position. + var invalidArg *types.InvalidArgumentException + if errors.As(err, &invalidArg) && + (pos.Type == types.ShardIteratorTypeAfterSequenceNumber || pos.Type == types.ShardIteratorTypeAtSequenceNumber) { + e.log.Warnf("Stored position for shard '%v' was rejected, falling back to the oldest retained record", e.shardID) + pos = types.StartingPosition{Type: types.ShardIteratorTypeTrimHorizon} + } + select { + case <-time.After(boff.NextBackOff()): + case <-e.ctx.Done(): + return + } + continue + } + } + + finished, continuation, sawEvent := e.consume(sub) + streamErr := sub.Err() + _ = sub.Close() + sub = nil + + // Progress, not a successful subscribe, resets the retry interval: 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. + if sawEvent { + boff.Reset() + } + + if finished { + e.finished.Store(true) + return + } + if e.ctx.Err() != nil { + return + } + if continuation != "" { + pos = types.StartingPosition{ + Type: types.ShardIteratorTypeAtSequenceNumber, + SequenceNumber: &continuation, + } + } + // Escalate the backoff whenever the subscription ended without + // delivering any event, whether or not it ended with an error: an + // errorless, eventless close is exactly how the subscription + // ping-pong from a shard-steal ends, and without escalating here it + // would resubscribe at the one-second floor forever, saturating the + // SubscribeToShard limit rather than backing off. + if streamErr != nil || !sawEvent { + if streamErr != nil { + e.log.Errorf("Enhanced fan-out subscription for shard '%v' failed: %v", e.shardID, streamErr) + } else { + e.log.Debugf("Enhanced fan-out subscription for shard '%v' ended without delivering any event", e.shardID) + } + select { + case <-time.After(boff.NextBackOff()): + case <-e.ctx.Done(): + return + } + } + } +} + +// consume reads events until the subscription ends. finished is true when the +// shard is closed and fully read (signalled by a nil continuation sequence). +// sawEvent reports whether the subscription delivered at least one shard event +// (with or without records), which the caller uses to decide whether the +// subscription made any progress before it ended. +func (e *efoRecordSource) consume(sub efoSubscription) (finished bool, continuation string, sawEvent bool) { + for { + select { + case ev, ok := <-sub.Events(): + if !ok { + return false, continuation, sawEvent + } + sev, ok := ev.(*types.SubscribeToShardEventStreamMemberSubscribeToShardEvent) + if !ok { + if ev != nil { + e.log.Errorf("Received unexpected event type: %T", ev) + } + continue + } + sawEvent = true + if len(sev.Value.Records) > 0 { + if !e.forward(sev.Value.Records) { + return false, continuation, sawEvent + } + } + if sev.Value.ContinuationSequenceNumber == nil { + return true, "", sawEvent + } + continuation = *sev.Value.ContinuationSequenceNumber + case <-e.ctx.Done(): + return false, continuation, sawEvent + } + } +} + +func (e *efoRecordSource) forward(recs []types.Record) bool { + select { + case e.recordsChan <- recs: + return true + case <-e.ctx.Done(): + return false + } +} + +// Fetch waits (bounded by fetchTimeout) for the next pushed batch, so that +// the calling consumer loop keeps servicing its commit timer. +func (e *efoRecordSource) Fetch(ctx context.Context) ([]types.Record, bool, error) { + timer := time.NewTimer(e.fetchTimeout) + defer timer.Stop() + select { + case recs, ok := <-e.recordsChan: + if !ok { + return nil, e.finished.Load(), nil + } + return recs, false, nil + case <-timer.C: + return nil, false, nil + case <-ctx.Done(): + return nil, false, ctx.Err() + } +} + +func (*efoRecordSource) Blocking() bool { return true } + +func (e *efoRecordSource) Close() { + e.cancel() + e.wg.Wait() +} diff --git a/internal/impl/aws/kinesis/input_efo_test.go b/internal/impl/aws/kinesis/input_efo_test.go new file mode 100644 index 0000000000..513175bba0 --- /dev/null +++ b/internal/impl/aws/kinesis/input_efo_test.go @@ -0,0 +1,702 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package kinesis + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/kinesis" + "github.com/aws/aws-sdk-go-v2/service/kinesis/types" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/redpanda-data/benthos/v4/public/service" +) + +type mockConsumerAPI struct { + describe func(ctx context.Context, in *kinesis.DescribeStreamConsumerInput, opts ...func(*kinesis.Options)) (*kinesis.DescribeStreamConsumerOutput, error) + register func(ctx context.Context, in *kinesis.RegisterStreamConsumerInput, opts ...func(*kinesis.Options)) (*kinesis.RegisterStreamConsumerOutput, error) +} + +func (m *mockConsumerAPI) DescribeStreamConsumer(ctx context.Context, in *kinesis.DescribeStreamConsumerInput, opts ...func(*kinesis.Options)) (*kinesis.DescribeStreamConsumerOutput, error) { + return m.describe(ctx, in, opts...) +} + +func (m *mockConsumerAPI) RegisterStreamConsumer(ctx context.Context, in *kinesis.RegisterStreamConsumerInput, opts ...func(*kinesis.Options)) (*kinesis.RegisterStreamConsumerOutput, error) { + return m.register(ctx, in, opts...) +} + +func fastEFOWaits(t *testing.T) { + t.Helper() + oldInterval := efoConsumerPollInterval + efoConsumerPollInterval = time.Millisecond + t.Cleanup(func() { + efoConsumerPollInterval = oldInterval + }) +} + +func TestEnsureEFOConsumerExistingActive(t *testing.T) { + fastEFOWaits(t) + api := &mockConsumerAPI{ + describe: func(_ context.Context, in *kinesis.DescribeStreamConsumerInput, _ ...func(*kinesis.Options)) (*kinesis.DescribeStreamConsumerOutput, error) { + assert.Equal(t, "my-app", *in.ConsumerName) + assert.Equal(t, "stream-arn", *in.StreamARN) + return &kinesis.DescribeStreamConsumerOutput{ + ConsumerDescription: &types.ConsumerDescription{ + ConsumerARN: aws.String("consumer-arn"), + ConsumerStatus: types.ConsumerStatusActive, + }, + }, nil + }, + register: func(_ context.Context, _ *kinesis.RegisterStreamConsumerInput, _ ...func(*kinesis.Options)) (*kinesis.RegisterStreamConsumerOutput, error) { + t.Fatal("register must not be called for an existing consumer") + return nil, nil + }, + } + + arn, err := ensureEFOConsumer(t.Context(), api, "stream-arn", "my-app", time.Second, service.MockResources().Logger()) + require.NoError(t, err) + assert.Equal(t, "consumer-arn", arn) +} + +func TestEnsureEFOConsumerRegistersMissing(t *testing.T) { + fastEFOWaits(t) + registered := false + describes := 0 + api := &mockConsumerAPI{ + describe: func(_ context.Context, _ *kinesis.DescribeStreamConsumerInput, _ ...func(*kinesis.Options)) (*kinesis.DescribeStreamConsumerOutput, error) { + describes++ + if !registered || describes < 3 { + return nil, &types.ResourceNotFoundException{} + } + return &kinesis.DescribeStreamConsumerOutput{ + ConsumerDescription: &types.ConsumerDescription{ + ConsumerARN: aws.String("new-arn"), + ConsumerStatus: types.ConsumerStatusActive, + }, + }, nil + }, + register: func(_ context.Context, in *kinesis.RegisterStreamConsumerInput, _ ...func(*kinesis.Options)) (*kinesis.RegisterStreamConsumerOutput, error) { + assert.Equal(t, "my-app", *in.ConsumerName) + registered = true + return &kinesis.RegisterStreamConsumerOutput{Consumer: &types.Consumer{ + ConsumerARN: aws.String("new-arn"), + ConsumerStatus: types.ConsumerStatusCreating, + }}, nil + }, + } + + arn, err := ensureEFOConsumer(t.Context(), api, "stream-arn", "my-app", time.Second, service.MockResources().Logger()) + require.NoError(t, err) + assert.Equal(t, "new-arn", arn) + assert.True(t, registered) +} + +func TestEnsureEFOConsumerWaitsForCreating(t *testing.T) { + fastEFOWaits(t) + describes := 0 + api := &mockConsumerAPI{ + describe: func(_ context.Context, _ *kinesis.DescribeStreamConsumerInput, _ ...func(*kinesis.Options)) (*kinesis.DescribeStreamConsumerOutput, error) { + describes++ + status := types.ConsumerStatusCreating + if describes >= 3 { + status = types.ConsumerStatusActive + } + return &kinesis.DescribeStreamConsumerOutput{ + ConsumerDescription: &types.ConsumerDescription{ + ConsumerARN: aws.String("consumer-arn"), + ConsumerStatus: status, + }, + }, nil + }, + } + + arn, err := ensureEFOConsumer(t.Context(), api, "stream-arn", "my-app", time.Second, service.MockResources().Logger()) + require.NoError(t, err) + assert.Equal(t, "consumer-arn", arn) + assert.GreaterOrEqual(t, describes, 3) +} + +func TestEnsureEFOConsumerPermissionError(t *testing.T) { + fastEFOWaits(t) + api := &mockConsumerAPI{ + describe: func(_ context.Context, _ *kinesis.DescribeStreamConsumerInput, _ ...func(*kinesis.Options)) (*kinesis.DescribeStreamConsumerOutput, error) { + return nil, errors.New("AccessDeniedException") + }, + } + + _, err := ensureEFOConsumer(t.Context(), api, "stream-arn", "my-app", time.Second, service.MockResources().Logger()) + require.Error(t, err) + assert.Contains(t, err.Error(), "kinesis:DescribeStreamConsumer") +} + +func TestEnsureEFOConsumerConcurrentRegistration(t *testing.T) { + fastEFOWaits(t) + describes := 0 + api := &mockConsumerAPI{ + describe: func(_ context.Context, _ *kinesis.DescribeStreamConsumerInput, _ ...func(*kinesis.Options)) (*kinesis.DescribeStreamConsumerOutput, error) { + describes++ + if describes == 1 { + return nil, &types.ResourceNotFoundException{} + } + return &kinesis.DescribeStreamConsumerOutput{ + ConsumerDescription: &types.ConsumerDescription{ + ConsumerARN: aws.String("consumer-arn"), + ConsumerStatus: types.ConsumerStatusActive, + }, + }, nil + }, + register: func(_ context.Context, _ *kinesis.RegisterStreamConsumerInput, _ ...func(*kinesis.Options)) (*kinesis.RegisterStreamConsumerOutput, error) { + // Another instance won the race. + return nil, &types.ResourceInUseException{} + }, + } + + arn, err := ensureEFOConsumer(t.Context(), api, "stream-arn", "my-app", time.Second, service.MockResources().Logger()) + require.NoError(t, err) + assert.Equal(t, "consumer-arn", arn) +} + +func TestEnsureEFOConsumerDeletingFails(t *testing.T) { + fastEFOWaits(t) + api := &mockConsumerAPI{ + describe: func(_ context.Context, _ *kinesis.DescribeStreamConsumerInput, _ ...func(*kinesis.Options)) (*kinesis.DescribeStreamConsumerOutput, error) { + return &kinesis.DescribeStreamConsumerOutput{ + ConsumerDescription: &types.ConsumerDescription{ + ConsumerARN: aws.String("consumer-arn"), + ConsumerStatus: types.ConsumerStatusDeleting, + }, + }, nil + }, + } + + _, err := ensureEFOConsumer(t.Context(), api, "stream-arn", "my-app", time.Second, service.MockResources().Logger()) + require.Error(t, err) + assert.Contains(t, err.Error(), "deleted") +} + +func TestEnsureEFOConsumerNilDescription(t *testing.T) { + fastEFOWaits(t) + attempts := 0 + api := &mockConsumerAPI{ + describe: func(_ context.Context, _ *kinesis.DescribeStreamConsumerInput, _ ...func(*kinesis.Options)) (*kinesis.DescribeStreamConsumerOutput, error) { + attempts++ + if attempts < 2 { + // First attempt: nil ConsumerDescription despite nil error (quirky API response) + return &kinesis.DescribeStreamConsumerOutput{ + ConsumerDescription: nil, + }, nil + } + // Subsequent attempts: normal response with ACTIVE consumer + return &kinesis.DescribeStreamConsumerOutput{ + ConsumerDescription: &types.ConsumerDescription{ + ConsumerARN: aws.String("consumer-arn"), + ConsumerStatus: types.ConsumerStatusActive, + }, + }, nil + }, + } + + arn, err := ensureEFOConsumer(t.Context(), api, "stream-arn", "my-app", time.Second, service.MockResources().Logger()) + require.NoError(t, err) + assert.Equal(t, "consumer-arn", arn) + assert.GreaterOrEqual(t, attempts, 2) +} + +func TestEnsureEFOConsumerActiveWithNilARN(t *testing.T) { + fastEFOWaits(t) + api := &mockConsumerAPI{ + describe: func(_ context.Context, _ *kinesis.DescribeStreamConsumerInput, _ ...func(*kinesis.Options)) (*kinesis.DescribeStreamConsumerOutput, error) { + return &kinesis.DescribeStreamConsumerOutput{ + ConsumerDescription: &types.ConsumerDescription{ + ConsumerARN: nil, // quirky API response: ACTIVE but no ARN + ConsumerStatus: types.ConsumerStatusActive, + }, + }, nil + }, + } + + _, err := ensureEFOConsumer(t.Context(), api, "stream-arn", "my-app", time.Second, service.MockResources().Logger()) + require.Error(t, err) + assert.Contains(t, err.Error(), "nil ARN") +} + +type fakeSubscription struct { + events chan types.SubscribeToShardEventStream + err error + closed chan struct{} +} + +func newFakeSubscription() *fakeSubscription { + return &fakeSubscription{ + events: make(chan types.SubscribeToShardEventStream, 16), + closed: make(chan struct{}), + } +} + +func (f *fakeSubscription) Events() <-chan types.SubscribeToShardEventStream { return f.events } + +func (f *fakeSubscription) Close() error { + select { + case <-f.closed: + default: + close(f.closed) + } + return nil +} + +func (f *fakeSubscription) Err() error { return f.err } + +func (f *fakeSubscription) send(continuation string, recs ...types.Record) { + f.events <- &types.SubscribeToShardEventStreamMemberSubscribeToShardEvent{ + Value: types.SubscribeToShardEvent{ + ContinuationSequenceNumber: aws.String(continuation), + MillisBehindLatest: aws.Int64(0), + Records: recs, + }, + } +} + +// sendFinal emits the closed-shard terminator (nil continuation). +func (f *fakeSubscription) sendFinal(recs ...types.Record) { + f.events <- &types.SubscribeToShardEventStreamMemberSubscribeToShardEvent{ + Value: types.SubscribeToShardEvent{ + MillisBehindLatest: aws.Int64(0), + Records: recs, + }, + } +} + +func rec(seq string) types.Record { + return types.Record{SequenceNumber: aws.String(seq), Data: []byte(seq)} +} + +// fastEFOResubscribe shrinks the resubscribe floor (and therefore the +// backoff's initial interval) so tests that exercise resubscription don't +// have to wait out the real one-second-per-shard-per-consumer API limit. +func fastEFOResubscribe(t *testing.T) { + t.Helper() + old := efoResubscribeFloor + efoResubscribeFloor = time.Millisecond + t.Cleanup(func() { + efoResubscribeFloor = old + }) +} + +func TestEFOSourceForwardsRecords(t *testing.T) { + sub := newFakeSubscription() + src, err := newEFORecordSource(t.Context(), func(_ context.Context, pos types.StartingPosition) (efoSubscription, error) { + assert.Equal(t, types.ShardIteratorTypeTrimHorizon, pos.Type) + return sub, nil + }, "shard-0", "", true, 100*time.Millisecond, 30*time.Second, service.MockResources().Logger()) + require.NoError(t, err) + defer src.Close() + + sub.send("c1", rec("1"), rec("2")) + + recs, done, err := src.Fetch(t.Context()) + require.NoError(t, err) + assert.False(t, done) + require.Len(t, recs, 2) + assert.True(t, src.Blocking()) +} + +func TestEFOSourceFetchTimesOutEmpty(t *testing.T) { + sub := newFakeSubscription() + src, err := newEFORecordSource(t.Context(), func(_ context.Context, _ types.StartingPosition) (efoSubscription, error) { + return sub, nil + }, "shard-0", "", true, 20*time.Millisecond, 30*time.Second, service.MockResources().Logger()) + require.NoError(t, err) + defer src.Close() + + recs, done, err := src.Fetch(t.Context()) + require.NoError(t, err) + assert.False(t, done) + assert.Empty(t, recs) +} + +func TestEFOSourceResubscribesAtContinuation(t *testing.T) { + fastEFOResubscribe(t) + + var mu sync.Mutex + var positions []types.StartingPosition + subs := make(chan *fakeSubscription, 2) + first, second := newFakeSubscription(), newFakeSubscription() + subs <- first + subs <- second + + src, err := newEFORecordSource(t.Context(), func(_ context.Context, pos types.StartingPosition) (efoSubscription, error) { + mu.Lock() + positions = append(positions, pos) + mu.Unlock() + return <-subs, nil + }, "shard-0", "start-seq", true, 100*time.Millisecond, 30*time.Second, service.MockResources().Logger()) + require.NoError(t, err) + defer src.Close() + + first.send("cont-1", rec("1")) + close(first.events) // AWS ends the ~5 minute subscription + + second.send("cont-2", rec("2")) + + var got []string + for len(got) < 2 { + recs, done, err := src.Fetch(t.Context()) + require.NoError(t, err) + require.False(t, done) + for _, r := range recs { + got = append(got, *r.SequenceNumber) + } + } + assert.Equal(t, []string{"1", "2"}, got) + + mu.Lock() + defer mu.Unlock() + require.Len(t, positions, 2) + assert.Equal(t, types.ShardIteratorTypeAfterSequenceNumber, positions[0].Type) + assert.Equal(t, "start-seq", *positions[0].SequenceNumber) + assert.Equal(t, types.ShardIteratorTypeAtSequenceNumber, positions[1].Type) + assert.Equal(t, "cont-1", *positions[1].SequenceNumber) +} + +func TestEFOSourceStreamErrorResubscribes(t *testing.T) { + fastEFOResubscribe(t) + + var mu sync.Mutex + var positions []types.StartingPosition + subs := make(chan *fakeSubscription, 2) + first, second := newFakeSubscription(), newFakeSubscription() + first.err = errors.New("stream error") + subs <- first + subs <- second + + src, err := newEFORecordSource(t.Context(), func(_ context.Context, pos types.StartingPosition) (efoSubscription, error) { + mu.Lock() + positions = append(positions, pos) + mu.Unlock() + return <-subs, nil + }, "shard-0", "", true, 100*time.Millisecond, 30*time.Second, service.MockResources().Logger()) + require.NoError(t, err) + defer src.Close() + + first.send("cont-1", rec("1")) + close(first.events) // stream ends with a non-nil Err() + + second.send("cont-2", rec("2")) + + var got []string + for len(got) < 2 { + recs, done, err := src.Fetch(t.Context()) + require.NoError(t, err) + require.False(t, done) + for _, r := range recs { + got = append(got, *r.SequenceNumber) + } + } + assert.Equal(t, []string{"1", "2"}, got) + + mu.Lock() + defer mu.Unlock() + require.Len(t, positions, 2) + assert.Equal(t, types.ShardIteratorTypeAtSequenceNumber, positions[1].Type) + assert.Equal(t, "cont-1", *positions[1].SequenceNumber) +} + +func TestEFOSourceAnchorsTimestampWhenNotOldest(t *testing.T) { + sub := newFakeSubscription() + before := time.Now() + src, err := newEFORecordSource(t.Context(), func(_ context.Context, pos types.StartingPosition) (efoSubscription, error) { + assert.Equal(t, types.ShardIteratorTypeAtTimestamp, pos.Type) + if assert.NotNil(t, pos.Timestamp) { + assert.False(t, pos.Timestamp.Before(before)) + } + return sub, nil + }, "shard-0", "", false, 100*time.Millisecond, 30*time.Second, service.MockResources().Logger()) + require.NoError(t, err) + defer src.Close() +} + +func TestEFOSourceShardClosed(t *testing.T) { + sub := newFakeSubscription() + src, err := newEFORecordSource(t.Context(), func(_ context.Context, _ types.StartingPosition) (efoSubscription, error) { + return sub, nil + }, "shard-0", "", true, 100*time.Millisecond, 30*time.Second, service.MockResources().Logger()) + require.NoError(t, err) + defer src.Close() + + sub.sendFinal(rec("1")) + + recs, done, err := src.Fetch(t.Context()) + require.NoError(t, err) + assert.False(t, done) + require.Len(t, recs, 1) + + // The next fetch reports the closed shard. + deadline := time.Now().Add(2 * time.Second) + for { + recs, done, err = src.Fetch(t.Context()) + require.NoError(t, err) + assert.Empty(t, recs) + if done || time.Now().After(deadline) { + break + } + } + assert.True(t, done) +} + +func TestEFOSourceInitialSubscribeErrorFailsFast(t *testing.T) { + _, err := newEFORecordSource(t.Context(), func(_ context.Context, _ types.StartingPosition) (efoSubscription, error) { + return nil, errors.New("AccessDeniedException") + }, "shard-0", "", true, 100*time.Millisecond, 30*time.Second, service.MockResources().Logger()) + require.Error(t, err) +} + +func TestEFOSourceInitialSubscribeNonResourceInUseFailsFastWithOneCall(t *testing.T) { + fastEFOResubscribe(t) + + calls := 0 + _, err := newEFORecordSource(t.Context(), func(_ context.Context, _ types.StartingPosition) (efoSubscription, error) { + calls++ + return nil, errors.New("AccessDenied") + }, "shard-0", "", true, 100*time.Millisecond, 30*time.Second, service.MockResources().Logger()) + require.Error(t, err) + assert.Equal(t, 1, calls) +} + +func TestEFOSourceInvalidSequenceFallsBackToConfiguredStart(t *testing.T) { + sub := newFakeSubscription() + var positions []types.StartingPosition + calls := 0 + src, err := newEFORecordSource(t.Context(), func(_ context.Context, pos types.StartingPosition) (efoSubscription, error) { + positions = append(positions, pos) + calls++ + if calls == 1 { + return nil, &types.InvalidArgumentException{} + } + return sub, nil + }, "shard-0", "stale-seq", true, 100*time.Millisecond, 30*time.Second, service.MockResources().Logger()) + require.NoError(t, err) + defer src.Close() + + require.Len(t, positions, 2) + assert.Equal(t, types.ShardIteratorTypeAfterSequenceNumber, positions[0].Type) + require.NotNil(t, positions[0].SequenceNumber) + assert.Equal(t, "stale-seq", *positions[0].SequenceNumber) + assert.Equal(t, types.ShardIteratorTypeTrimHorizon, positions[1].Type) +} + +func TestEFOSourceInvalidSequenceFallsBackToTrimHorizonWhenNotOldest(t *testing.T) { + sub := newFakeSubscription() + var positions []types.StartingPosition + calls := 0 + src, err := newEFORecordSource(t.Context(), func(_ context.Context, pos types.StartingPosition) (efoSubscription, error) { + positions = append(positions, pos) + calls++ + if calls == 1 { + return nil, &types.InvalidArgumentException{} + } + return sub, nil + }, "shard-0", "stale-seq", false, 100*time.Millisecond, 30*time.Second, service.MockResources().Logger()) + require.NoError(t, err) + defer src.Close() + + require.Len(t, positions, 2) + assert.Equal(t, types.ShardIteratorTypeAfterSequenceNumber, positions[0].Type) + // This shard demonstrably had a committed position, so resuming at "now" + // would silently skip every record still retained ahead of the expired + // sequence: the fallback must be TRIM_HORIZON regardless of + // start_from_oldest. + assert.Equal(t, types.ShardIteratorTypeTrimHorizon, positions[1].Type) + assert.Nil(t, positions[1].Timestamp) +} + +func TestEFOSourcePumpFallsBackOnInvalidSequence(t *testing.T) { + fastEFOResubscribe(t) + + var mu sync.Mutex + var positions []types.StartingPosition + sub := newFakeSubscription() + + src, err := newEFORecordSource(t.Context(), func(_ context.Context, pos types.StartingPosition) (efoSubscription, error) { + mu.Lock() + positions = append(positions, pos) + n := len(positions) + mu.Unlock() + switch n { + case 1: + // The shard's previous owner still holds the one allowed + // subscription, so the pump takes over the retry loop. + return nil, &types.ResourceInUseException{} + case 2: + // The stored sequence has aged out of the retention window. + return nil, &types.InvalidArgumentException{} + } + return sub, nil + }, "shard-0", "stale-seq", false, 20*time.Millisecond, 30*time.Second, service.MockResources().Logger()) + require.NoError(t, err) + defer src.Close() + + sub.send("c1", rec("1")) + + var got int + deadline := time.Now().Add(5 * time.Second) + for got == 0 && time.Now().Before(deadline) { + recs, _, err := src.Fetch(t.Context()) + require.NoError(t, err) + got = len(recs) + } + require.Equal(t, 1, got) + + mu.Lock() + defer mu.Unlock() + require.GreaterOrEqual(t, len(positions), 3) + assert.Equal(t, types.ShardIteratorTypeAfterSequenceNumber, positions[0].Type) + assert.Equal(t, types.ShardIteratorTypeAfterSequenceNumber, positions[1].Type) + // Without the fallback the pump would retry the rejected sequence forever. + assert.Equal(t, types.ShardIteratorTypeTrimHorizon, positions[2].Type) +} + +func TestEFOSourceOtherErrorWithSequenceDoesNotFallBack(t *testing.T) { + calls := 0 + _, err := newEFORecordSource(t.Context(), func(_ context.Context, _ types.StartingPosition) (efoSubscription, error) { + calls++ + return nil, errors.New("AccessDeniedException") + }, "shard-0", "stale-seq", true, 100*time.Millisecond, 30*time.Second, service.MockResources().Logger()) + require.Error(t, err) + assert.Equal(t, 1, calls) +} + +func TestEFOSourceInitialSubscribeResourceInUseDoesNotBlock(t *testing.T) { + fastEFOResubscribe(t) + + const failCount = 3 + var calls atomic.Int32 + sub := newFakeSubscription() + + start := time.Now() + src, err := newEFORecordSource(t.Context(), func(_ context.Context, _ types.StartingPosition) (efoSubscription, error) { + n := calls.Add(1) + if n <= failCount { + return nil, &types.ResourceInUseException{} + } + return sub, nil + }, "shard-0", "", true, 100*time.Millisecond, 30*time.Second, service.MockResources().Logger()) + require.NoError(t, err) + defer src.Close() + + // The constructor must not block synchronously retrying/waiting out the + // ResourceInUseException; it should return as soon as the pump goroutine + // is started. + assert.Less(t, time.Since(start), time.Second) + + // Once the subscribe fn stops failing, the background pump picks it up + // and records flow normally. + sub.send("c1", rec("1")) + + recs, done, err := src.Fetch(t.Context()) + require.NoError(t, err) + assert.False(t, done) + require.Len(t, recs, 1) + + assert.GreaterOrEqual(t, calls.Load(), int32(failCount+1)) +} + +func TestEFOSourceCloseStopsPump(t *testing.T) { + sub := newFakeSubscription() + src, err := newEFORecordSource(t.Context(), func(_ context.Context, _ types.StartingPosition) (efoSubscription, error) { + return sub, nil + }, "shard-0", "", true, 100*time.Millisecond, 30*time.Second, service.MockResources().Logger()) + require.NoError(t, err) + + src.Close() // must not hang + select { + case <-sub.closed: + default: + t.Fatal("expected the subscription to be closed") + } +} + +// TestEFOSourceEscalatesBackoffWithoutEvents covers the case where a +// subscription's event stream closes without ever delivering an event and +// without sub.Err() being set (exactly how the subscription ping-pong from a +// shard steal ends). Prior to the fix, boff was neither reset (sawEvent is +// false) nor advanced (the escalation was gated on a non-nil stream error), +// so the pump would resubscribe forever at the bare efoResubscribeFloor +// cadence instead of escalating. +func TestEFOSourceEscalatesBackoffWithoutEvents(t *testing.T) { + // A larger-than-1ms floor is used (rather than the fastEFOResubscribe + // helper's 1ms) so that the escalating backoff dominates measured gaps + // over fixed scheduling/timer overhead, keeping the elapsed-time + // assertion below deterministic rather than a tight, flake-prone bound. + old := efoResubscribeFloor + efoResubscribeFloor = 5 * time.Millisecond + t.Cleanup(func() { + efoResubscribeFloor = old + }) + + const wantSubscribes = 10 + + var ( + mu sync.Mutex + times []time.Time + ) + allDone := make(chan struct{}) + + src, err := newEFORecordSource(t.Context(), func(_ context.Context, _ types.StartingPosition) (efoSubscription, error) { + sub := newFakeSubscription() + close(sub.events) // ends immediately: no events, no error + + mu.Lock() + times = append(times, time.Now()) + n := len(times) + mu.Unlock() + + if n == wantSubscribes { + close(allDone) + } + return sub, nil + }, "shard-0", "", true, 20*time.Millisecond, 30*time.Second, service.MockResources().Logger()) + require.NoError(t, err) + defer src.Close() + + select { + case <-allDone: + case <-time.After(3 * time.Second): + } + + mu.Lock() + got := append([]time.Time{}, times...) + mu.Unlock() + + require.GreaterOrEqual(t, len(got), wantSubscribes, + "expected %d subscribe cycles within the deadline, got %d", wantSubscribes, len(got)) + + // Without escalation, every resubscribe would be spaced by roughly + // efoResubscribeFloor alone, so the total elapsed time across all the + // cycles would sit close to (n-1)*efoResubscribeFloor. With escalation, + // the exponentially growing backoff wait dominates, so require the + // actual elapsed time to clear that floor-only baseline by a wide, + // noise-tolerant margin. + elapsed := got[len(got)-1].Sub(got[0]) + floorOnlyBaseline := time.Duration(len(got)-1) * efoResubscribeFloor + assert.Greater(t, elapsed, floorOnlyBaseline*3, + "expected resubscribe backoff to escalate when no events are delivered: elapsed=%v floorOnlyBaseline=%v", elapsed, floorOnlyBaseline) +} diff --git a/internal/impl/aws/kinesis/input_record_source.go b/internal/impl/aws/kinesis/input_record_source.go new file mode 100644 index 0000000000..5e34bb51cf --- /dev/null +++ b/internal/impl/aws/kinesis/input_record_source.go @@ -0,0 +1,214 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package kinesis + +import ( + "context" + "errors" + "time" + + "github.com/aws/aws-sdk-go-v2/service/kinesis" + "github.com/aws/aws-sdk-go-v2/service/kinesis/types" + + "github.com/redpanda-data/benthos/v4/public/service" +) + +// shardRecordSource yields batches of records from a single Kinesis shard, +// abstracting over the polling (GetRecords) and enhanced fan-out +// (SubscribeToShard) consumption models. +type shardRecordSource interface { + // Fetch returns the next batch of records. done is true once the shard is + // closed and fully consumed. A blocking source waits internally (bounded) + // for data; a non-blocking source returns immediately and relies on the + // caller to pace retries. + Fetch(ctx context.Context) (recs []types.Record, done bool, err error) + // Blocking reports whether Fetch waits for data internally, in which case + // the caller must not add its own backoff to empty results. + Blocking() bool + // Close releases any underlying resources. + Close() +} + +// errPollGateWaiting signals that a Fetch returned early because the +// poll_period gate has not yet elapsed, rather than because the shard +// had no records. The consumer loop retries immediately without arming +// its failure backoff; the gate itself is the pacing mechanism. +var errPollGateWaiting = errors.New("poll gate waiting") + +// kinesisPollAPI is the subset of the Kinesis API used by the polling source. +type kinesisPollAPI interface { + GetShardIterator(ctx context.Context, params *kinesis.GetShardIteratorInput, optFns ...func(*kinesis.Options)) (*kinesis.GetShardIteratorOutput, error) + GetRecords(ctx context.Context, params *kinesis.GetRecordsInput, optFns ...func(*kinesis.Options)) (*kinesis.GetRecordsOutput, error) +} + +// pollingRecordSource consumes a shard via GetRecords, owning the shard +// iterator, its refresh on expiry, and the optional poll_period gate. +type pollingRecordSource struct { + api kinesisPollAPI + streamARN string + shardID string + startFromOldest bool + pollPeriod time.Duration + // maxGateWait bounds how long a single Fetch may sleep inside the + // poll_period gate before handing control back to the caller, so that a + // long poll_period cannot starve the consumer loop's commit timer. A value + // of zero leaves the gate wait uncapped. + maxGateWait time.Duration + sequenceFn func() string + log *service.Logger + + iter string + lastPoll time.Time +} + +func newPollingRecordSource(ctx context.Context, api kinesisPollAPI, streamARN, shardID, startingSequence string, startFromOldest bool, pollPeriod, maxGateWait time.Duration, sequenceFn func() string, log *service.Logger) (*pollingRecordSource, error) { + p := &pollingRecordSource{ + api: api, + streamARN: streamARN, + shardID: shardID, + startFromOldest: startFromOldest, + pollPeriod: pollPeriod, + maxGateWait: maxGateWait, + sequenceFn: sequenceFn, + log: log, + } + iter, err := p.getIter(ctx, startingSequence) + if err != nil { + return nil, err + } + p.iter = iter + return p, nil +} + +func (p *pollingRecordSource) getIter(ctx context.Context, sequence string) (string, error) { + iterType := types.ShardIteratorTypeTrimHorizon + if !p.startFromOldest { + iterType = types.ShardIteratorTypeLatest + } + var startingSequence *string + if sequence != "" { + iterType = types.ShardIteratorTypeAfterSequenceNumber + startingSequence = &sequence + } + + res, err := p.api.GetShardIterator(ctx, &kinesis.GetShardIteratorInput{ + StreamARN: &p.streamARN, + ShardId: &p.shardID, + StartingSequenceNumber: startingSequence, + ShardIteratorType: iterType, + }) + if err != nil { + // A sequence that has aged out of the shard's retention window is + // rejected outright rather than yielding an empty iterator, and no + // number of retries will make it resolvable again, so fall through to + // the TRIM_HORIZON fallback below. + var invalidArg *types.InvalidArgumentException + if startingSequence == nil || !errors.As(err, &invalidArg) { + return "", err + } + p.log.Warnf("Stored sequence for shard '%v' was rejected, falling back to the oldest retained record", p.shardID) + } + + var iter string + if res != nil && res.ShardIterator != nil { + iter = *res.ShardIterator + } + if iter == "" { + // If we failed to obtain from a sequence we start from beginning + iterType = types.ShardIteratorTypeTrimHorizon + + res, err := p.api.GetShardIterator(ctx, &kinesis.GetShardIteratorInput{ + StreamARN: &p.streamARN, + ShardId: &p.shardID, + ShardIteratorType: iterType, + }) + if err != nil { + return "", err + } + + if res.ShardIterator != nil { + iter = *res.ShardIterator + } + } + if iter == "" { + return "", errors.New("obtaining shard iterator") + } + return iter, nil +} + +// Fetch pulls the next batch via GetRecords. The shard iterator is only +// replaced on success or an internal refresh, so a failed call can always be +// retried with the retained iterator. +// +// When the poll_period gate has not yet elapsed, Fetch sleeps for at most +// maxGateWait and then returns errPollGateWaiting without polling, leaving +// lastPoll untouched. The gate therefore still enforces the full minimum +// spacing between GetRecords calls (lastPoll only advances when GetRecords is +// actually invoked), whilst the caller stays free to service its commit timer +// and flush pending messages. errPollGateWaiting tells the caller this is +// gate pacing rather than an empty shard, so it retries immediately instead +// of arming its failure backoff. +func (p *pollingRecordSource) Fetch(ctx context.Context) ([]types.Record, bool, error) { + 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, errPollGateWaiting + } + } + } + p.lastPoll = time.Now() + + res, err := p.api.GetRecords(ctx, &kinesis.GetRecordsInput{ + StreamARN: &p.streamARN, + Limit: &awsKinesisDefaultLimit, + ShardIterator: &p.iter, + }) + if err != nil { + var aerr *types.ExpiredIteratorException + if errors.As(err, &aerr) { + p.log.Warn("Shard iterator expired, attempting to refresh") + newIter, ierr := p.getIter(ctx, p.sequenceFn()) + if ierr != nil { + p.log.Errorf("Failed to refresh shard iterator: %v", ierr) + } else { + p.iter = newIter + } + // Treated as an empty result so the caller applies its usual + // empty-result pacing, matching the pre-refactor behaviour. + return nil, false, nil + } + return nil, false, err + } + + nextIter := "" + if res.NextShardIterator != nil { + nextIter = *res.NextShardIterator + } + p.iter = nextIter + return res.Records, nextIter == "", nil +} + +func (*pollingRecordSource) Blocking() bool { return false } + +func (*pollingRecordSource) Close() {} diff --git a/internal/impl/aws/kinesis/input_record_source_test.go b/internal/impl/aws/kinesis/input_record_source_test.go new file mode 100644 index 0000000000..536e86dc3f --- /dev/null +++ b/internal/impl/aws/kinesis/input_record_source_test.go @@ -0,0 +1,344 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package kinesis + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/kinesis" + "github.com/aws/aws-sdk-go-v2/service/kinesis/types" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/redpanda-data/benthos/v4/public/service" +) + +type mockPollAPI struct { + getShardIterator func(ctx context.Context, in *kinesis.GetShardIteratorInput, opts ...func(*kinesis.Options)) (*kinesis.GetShardIteratorOutput, error) + getRecords func(ctx context.Context, in *kinesis.GetRecordsInput, opts ...func(*kinesis.Options)) (*kinesis.GetRecordsOutput, error) +} + +func (m *mockPollAPI) GetShardIterator(ctx context.Context, in *kinesis.GetShardIteratorInput, opts ...func(*kinesis.Options)) (*kinesis.GetShardIteratorOutput, error) { + return m.getShardIterator(ctx, in, opts...) +} + +func (m *mockPollAPI) GetRecords(ctx context.Context, in *kinesis.GetRecordsInput, opts ...func(*kinesis.Options)) (*kinesis.GetRecordsOutput, error) { + return m.getRecords(ctx, in, opts...) +} + +func staticIterAPI(iter string) *mockPollAPI { + return &mockPollAPI{ + getShardIterator: func(_ context.Context, _ *kinesis.GetShardIteratorInput, _ ...func(*kinesis.Options)) (*kinesis.GetShardIteratorOutput, error) { + return &kinesis.GetShardIteratorOutput{ShardIterator: aws.String(iter)}, nil + }, + } +} + +func TestPollingSourceFetchAdvancesIterator(t *testing.T) { + api := staticIterAPI("iter-1") + var gotIters []string + api.getRecords = func(_ context.Context, in *kinesis.GetRecordsInput, _ ...func(*kinesis.Options)) (*kinesis.GetRecordsOutput, error) { + gotIters = append(gotIters, *in.ShardIterator) + return &kinesis.GetRecordsOutput{ + Records: []types.Record{{SequenceNumber: aws.String("1")}}, + NextShardIterator: aws.String("iter-2"), + }, nil + } + + src, err := newPollingRecordSource(t.Context(), api, "arn", "shard-0", "", true, 0, time.Second, func() string { return "" }, service.MockResources().Logger()) + require.NoError(t, err) + + recs, done, err := src.Fetch(t.Context()) + require.NoError(t, err) + assert.False(t, done) + require.Len(t, recs, 1) + + _, _, err = src.Fetch(t.Context()) + require.NoError(t, err) + assert.Equal(t, []string{"iter-1", "iter-2"}, gotIters) + assert.False(t, src.Blocking()) +} + +func TestPollingSourceEndOfShard(t *testing.T) { + api := staticIterAPI("iter-1") + api.getRecords = func(_ context.Context, _ *kinesis.GetRecordsInput, _ ...func(*kinesis.Options)) (*kinesis.GetRecordsOutput, error) { + return &kinesis.GetRecordsOutput{ + Records: []types.Record{{SequenceNumber: aws.String("1")}}, + NextShardIterator: nil, // closed shard + }, nil + } + + src, err := newPollingRecordSource(t.Context(), api, "arn", "shard-0", "", true, 0, time.Second, func() string { return "" }, service.MockResources().Logger()) + require.NoError(t, err) + + recs, done, err := src.Fetch(t.Context()) + require.NoError(t, err) + assert.True(t, done) + assert.Len(t, recs, 1) +} + +func TestPollingSourceErrorKeepsIterator(t *testing.T) { + api := staticIterAPI("iter-1") + calls := 0 + api.getRecords = func(_ context.Context, in *kinesis.GetRecordsInput, _ ...func(*kinesis.Options)) (*kinesis.GetRecordsOutput, error) { + calls++ + if calls == 1 { + return nil, errors.New("boom") + } + assert.Equal(t, "iter-1", *in.ShardIterator) + return &kinesis.GetRecordsOutput{NextShardIterator: aws.String("iter-2")}, nil + } + + src, err := newPollingRecordSource(t.Context(), api, "arn", "shard-0", "", true, 0, time.Second, func() string { return "" }, service.MockResources().Logger()) + require.NoError(t, err) + + _, done, err := src.Fetch(t.Context()) + require.Error(t, err) + assert.False(t, done) + + _, _, err = src.Fetch(t.Context()) + require.NoError(t, err) + assert.Equal(t, 2, calls) +} + +func TestPollingSourceExpiredIteratorRefreshes(t *testing.T) { + var iterRequests []*kinesis.GetShardIteratorInput + api := &mockPollAPI{} + api.getShardIterator = func(_ context.Context, in *kinesis.GetShardIteratorInput, _ ...func(*kinesis.Options)) (*kinesis.GetShardIteratorOutput, error) { + iterRequests = append(iterRequests, in) + return &kinesis.GetShardIteratorOutput{ShardIterator: aws.String("fresh-iter")}, nil + } + calls := 0 + api.getRecords = func(_ context.Context, in *kinesis.GetRecordsInput, _ ...func(*kinesis.Options)) (*kinesis.GetRecordsOutput, error) { + calls++ + if calls == 1 { + return nil, &types.ExpiredIteratorException{} + } + assert.Equal(t, "fresh-iter", *in.ShardIterator) + return &kinesis.GetRecordsOutput{NextShardIterator: aws.String("iter-2")}, nil + } + + src, err := newPollingRecordSource(t.Context(), api, "arn", "shard-0", "", true, 0, time.Second, func() string { return "acked-seq" }, service.MockResources().Logger()) + require.NoError(t, err) + + // Expired iterator: no records, no error, iterator refreshed internally. + recs, done, err := src.Fetch(t.Context()) + require.NoError(t, err) + assert.False(t, done) + assert.Empty(t, recs) + + _, _, err = src.Fetch(t.Context()) + require.NoError(t, err) + + // First request is the constructor's, second is the refresh which must + // resume after the latest acked sequence. + require.Len(t, iterRequests, 2) + assert.Equal(t, types.ShardIteratorTypeAfterSequenceNumber, iterRequests[1].ShardIteratorType) + assert.Equal(t, "acked-seq", *iterRequests[1].StartingSequenceNumber) +} + +func TestPollingSourceIterFallbackToTrimHorizon(t *testing.T) { + var iterTypes []types.ShardIteratorType + api := &mockPollAPI{} + api.getShardIterator = func(_ context.Context, in *kinesis.GetShardIteratorInput, _ ...func(*kinesis.Options)) (*kinesis.GetShardIteratorOutput, error) { + iterTypes = append(iterTypes, in.ShardIteratorType) + if in.ShardIteratorType == types.ShardIteratorTypeAfterSequenceNumber { + return &kinesis.GetShardIteratorOutput{}, nil // empty iterator triggers fallback + } + return &kinesis.GetShardIteratorOutput{ShardIterator: aws.String("iter-1")}, nil + } + + _, err := newPollingRecordSource(t.Context(), api, "arn", "shard-0", "old-seq", true, 0, time.Second, func() string { return "" }, service.MockResources().Logger()) + require.NoError(t, err) + assert.Equal(t, []types.ShardIteratorType{ + types.ShardIteratorTypeAfterSequenceNumber, + types.ShardIteratorTypeTrimHorizon, + }, iterTypes) +} + +func TestPollingSourceIterFallbackOnInvalidSequence(t *testing.T) { + var iterTypes []types.ShardIteratorType + api := &mockPollAPI{} + api.getShardIterator = func(_ context.Context, in *kinesis.GetShardIteratorInput, _ ...func(*kinesis.Options)) (*kinesis.GetShardIteratorOutput, error) { + iterTypes = append(iterTypes, in.ShardIteratorType) + if in.ShardIteratorType == types.ShardIteratorTypeAfterSequenceNumber { + // A sequence aged out of the retention window is rejected outright. + return nil, &types.InvalidArgumentException{} + } + return &kinesis.GetShardIteratorOutput{ShardIterator: aws.String("iter-1")}, nil + } + + src, err := newPollingRecordSource(t.Context(), api, "arn", "shard-0", "old-seq", true, 0, time.Second, func() string { return "" }, service.MockResources().Logger()) + require.NoError(t, err) + assert.Equal(t, "iter-1", src.iter) + assert.Equal(t, []types.ShardIteratorType{ + types.ShardIteratorTypeAfterSequenceNumber, + types.ShardIteratorTypeTrimHorizon, + }, iterTypes) +} + +func TestPollingSourceIterOtherErrorWithSequenceFails(t *testing.T) { + calls := 0 + api := &mockPollAPI{} + api.getShardIterator = func(_ context.Context, _ *kinesis.GetShardIteratorInput, _ ...func(*kinesis.Options)) (*kinesis.GetShardIteratorOutput, error) { + calls++ + return nil, errors.New("AccessDeniedException") + } + + _, err := newPollingRecordSource(t.Context(), api, "arn", "shard-0", "old-seq", true, 0, time.Second, func() string { return "" }, service.MockResources().Logger()) + require.Error(t, err) + assert.Equal(t, 1, calls) +} + +func TestPollingSourceIterInvalidArgumentWithoutSequenceFails(t *testing.T) { + calls := 0 + api := &mockPollAPI{} + api.getShardIterator = func(_ context.Context, _ *kinesis.GetShardIteratorInput, _ ...func(*kinesis.Options)) (*kinesis.GetShardIteratorOutput, error) { + calls++ + return nil, &types.InvalidArgumentException{} + } + + // Without a stored sequence there is nothing to fall back from, so the + // error must surface rather than triggering a second request. + _, err := newPollingRecordSource(t.Context(), api, "arn", "shard-0", "", true, 0, time.Second, func() string { return "" }, service.MockResources().Logger()) + require.Error(t, err) + assert.Equal(t, 1, calls) +} + +func TestPollingSourcePollPeriodGate(t *testing.T) { + api := staticIterAPI("iter-1") + var callTimes []time.Time + api.getRecords = func(_ context.Context, _ *kinesis.GetRecordsInput, _ ...func(*kinesis.Options)) (*kinesis.GetRecordsOutput, error) { + callTimes = append(callTimes, time.Now()) + return &kinesis.GetRecordsOutput{NextShardIterator: aws.String("iter-2")}, nil + } + + const period = 50 * time.Millisecond + src, err := newPollingRecordSource(t.Context(), api, "arn", "shard-0", "", true, period, time.Second, func() string { return "" }, service.MockResources().Logger()) + require.NoError(t, err) + + for range 3 { + _, _, err = src.Fetch(t.Context()) + require.NoError(t, err) + } + require.Len(t, callTimes, 3) + for i := 1; i < len(callTimes); i++ { + assert.GreaterOrEqual(t, callTimes[i].Sub(callTimes[i-1]), period) + } +} + +func TestPollingSourcePollPeriodGateCapsFetchWait(t *testing.T) { + api := staticIterAPI("iter-1") + var callTimes []time.Time + api.getRecords = func(_ context.Context, _ *kinesis.GetRecordsInput, _ ...func(*kinesis.Options)) (*kinesis.GetRecordsOutput, error) { + callTimes = append(callTimes, time.Now()) + return &kinesis.GetRecordsOutput{NextShardIterator: aws.String("iter-2")}, nil + } + + const ( + period = 200 * time.Millisecond + maxGateWait = 10 * time.Millisecond + ) + src, err := newPollingRecordSource(t.Context(), api, "arn", "shard-0", "", true, period, maxGateWait, func() string { return "" }, service.MockResources().Logger()) + require.NoError(t, err) + + _, _, err = src.Fetch(t.Context()) + require.NoError(t, err) + require.Len(t, callTimes, 1) + + // A fetch immediately after a successful one must hand control back within + // roughly maxGateWait, without polling, rather than sleeping out the + // remaining poll period. The capped wait is signalled via the + // errPollGateWaiting sentinel rather than a nil error, so the caller can + // tell this apart from a genuinely empty shard and skip arming its + // failure backoff. + start := time.Now() + recs, done, err := src.Fetch(t.Context()) + require.ErrorIs(t, err, errPollGateWaiting) + assert.False(t, done) + assert.Empty(t, recs) + assert.Len(t, callTimes, 1) + assert.Less(t, time.Since(start), period/2) + + // Repeated fetches keep yielding the sentinel until the full period has + // elapsed since the last GetRecords call. + deadline := time.Now().Add(5 * time.Second) + for len(callTimes) < 2 && time.Now().Before(deadline) { + _, _, err = src.Fetch(t.Context()) + if err != nil { + require.ErrorIs(t, err, errPollGateWaiting) + } + } + require.Len(t, callTimes, 2) + assert.GreaterOrEqual(t, callTimes[1].Sub(callTimes[0]), period) +} + +// TestPollingSourcePollPeriodGateSpacingUnderCap simulates the consumer +// loop's retry-immediately-on-sentinel behaviour (no backoff armed) by +// calling Fetch in a tight loop, and checks that GetRecords calls still land +// at roughly pollPeriod spacing rather than pollPeriod plus a failure +// backoff, even though almost every Fetch is capped well below the period. +func TestPollingSourcePollPeriodGateSpacingUnderCap(t *testing.T) { + api := staticIterAPI("iter-1") + var callTimes []time.Time + api.getRecords = func(_ context.Context, _ *kinesis.GetRecordsInput, _ ...func(*kinesis.Options)) (*kinesis.GetRecordsOutput, error) { + callTimes = append(callTimes, time.Now()) + return &kinesis.GetRecordsOutput{NextShardIterator: aws.String("iter-2")}, nil + } + + const ( + period = 100 * time.Millisecond + maxGateWait = 20 * time.Millisecond + ) + src, err := newPollingRecordSource(t.Context(), api, "arn", "shard-0", "", true, period, maxGateWait, func() string { return "" }, service.MockResources().Logger()) + require.NoError(t, err) + + const wantCalls = 4 + deadline := time.Now().Add(5 * time.Second) + for len(callTimes) < wantCalls && time.Now().Before(deadline) { + _, _, err = src.Fetch(t.Context()) + if err != nil { + require.ErrorIs(t, err, errPollGateWaiting) + } + } + require.GreaterOrEqual(t, len(callTimes), wantCalls) + + for i := 1; i < len(callTimes); i++ { + gap := callTimes[i].Sub(callTimes[i-1]) + assert.GreaterOrEqual(t, gap, period) + assert.Less(t, gap, time.Duration(float64(period)*1.5)) + } +} + +func TestPollingSourcePollPeriodZeroNoDelay(t *testing.T) { + api := staticIterAPI("iter-1") + api.getRecords = func(_ context.Context, _ *kinesis.GetRecordsInput, _ ...func(*kinesis.Options)) (*kinesis.GetRecordsOutput, error) { + return &kinesis.GetRecordsOutput{NextShardIterator: aws.String("iter-2")}, nil + } + + src, err := newPollingRecordSource(t.Context(), api, "arn", "shard-0", "", true, 0, time.Second, func() string { return "" }, service.MockResources().Logger()) + require.NoError(t, err) + + start := time.Now() + for range 10 { + _, _, err = src.Fetch(t.Context()) + require.NoError(t, err) + } + assert.Less(t, time.Since(start), 20*time.Millisecond) +} diff --git a/internal/impl/aws/kinesis/input_test.go b/internal/impl/aws/kinesis/input_test.go index b851f13d24..783670463c 100644 --- a/internal/impl/aws/kinesis/input_test.go +++ b/internal/impl/aws/kinesis/input_test.go @@ -16,9 +16,13 @@ package kinesis import ( "testing" + "time" + "github.com/aws/aws-sdk-go-v2/aws" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + + "github.com/redpanda-data/benthos/v4/public/service" ) func TestStreamIDParser(t *testing.T) { @@ -77,3 +81,280 @@ func TestStreamIDParser(t *testing.T) { }) } } + +func TestKinesisInputPollPeriodConfig(t *testing.T) { + pConf, err := kinesisInputSpec().ParseYAML(` +streams: [ foo ] +poll_period: 250ms +`, nil) + require.NoError(t, err) + + conf, err := kinesisInputConfigFromParsed(pConf) + require.NoError(t, err) + assert.Equal(t, 250*time.Millisecond, conf.PollPeriod) +} + +func TestKinesisInputPollPeriodDefault(t *testing.T) { + pConf, err := kinesisInputSpec().ParseYAML(` +streams: [ foo ] +`, nil) + require.NoError(t, err) + + conf, err := kinesisInputConfigFromParsed(pConf) + require.NoError(t, err) + assert.Equal(t, time.Duration(0), conf.PollPeriod) +} + +func TestKinesisInputPollPeriodExceedsLeasePeriodFails(t *testing.T) { + pConf, err := kinesisInputSpec().ParseYAML(` +streams: [ foo ] +poll_period: 60s +lease_period: 30s +`, nil) + require.NoError(t, err) + + conf, err := kinesisInputConfigFromParsed(pConf) + require.NoError(t, err) + + _, err = newKinesisReaderFromConfig(conf, service.BatchPolicy{}, aws.Config{}, aws.Config{}, service.MockResources()) + require.Error(t, err) + assert.Contains(t, err.Error(), "lease_period") +} + +func TestKinesisInputPollPeriodBetweenCommitAndLeasePeriodSucceeds(t *testing.T) { + pConf, err := kinesisInputSpec().ParseYAML(` +streams: [ foo ] +poll_period: 10s +commit_period: 5s +lease_period: 30s +`, nil) + require.NoError(t, err) + + conf, err := kinesisInputConfigFromParsed(pConf) + require.NoError(t, err) + + _, err = newKinesisReaderFromConfig(conf, service.BatchPolicy{}, aws.Config{}, aws.Config{}, service.MockResources()) + require.NoError(t, err) +} + +func TestKinesisInputEFOConfig(t *testing.T) { + pConf, err := kinesisInputSpec().ParseYAML(` +streams: [ foo ] +enhanced_fan_out: + enabled: true + consumer_name: my-app +`, nil) + require.NoError(t, err) + + conf, err := kinesisInputConfigFromParsed(pConf) + require.NoError(t, err) + assert.True(t, conf.EFOEnabled) + assert.Equal(t, "my-app", conf.EFOConsumerName) +} + +func TestKinesisInputEFODisabledByDefault(t *testing.T) { + pConf, err := kinesisInputSpec().ParseYAML(` +streams: [ foo ] +`, nil) + require.NoError(t, err) + + conf, err := kinesisInputConfigFromParsed(pConf) + require.NoError(t, err) + assert.False(t, conf.EFOEnabled) +} + +func TestKinesisInputEFOActivationTimeoutDefault(t *testing.T) { + pConf, err := kinesisInputSpec().ParseYAML(` +streams: [ foo ] +enhanced_fan_out: + enabled: true + consumer_name: my-app +`, nil) + require.NoError(t, err) + + conf, err := kinesisInputConfigFromParsed(pConf) + require.NoError(t, err) + assert.Equal(t, time.Minute, conf.EFOActivationTimeout) +} + +func TestKinesisInputEFOActivationTimeoutCustom(t *testing.T) { + pConf, err := kinesisInputSpec().ParseYAML(` +streams: [ foo ] +enhanced_fan_out: + enabled: true + consumer_name: my-app + consumer_activation_timeout: 90s +`, nil) + require.NoError(t, err) + + conf, err := kinesisInputConfigFromParsed(pConf) + require.NoError(t, err) + assert.Equal(t, 90*time.Second, conf.EFOActivationTimeout) +} + +func TestKinesisInputEFOActivationTimeoutZeroFails(t *testing.T) { + pConf, err := kinesisInputSpec().ParseYAML(` +streams: [ foo ] +enhanced_fan_out: + enabled: true + consumer_name: my-app + consumer_activation_timeout: 0s +`, nil) + require.NoError(t, err) + + _, err = kinesisInputConfigFromParsed(pConf) + require.Error(t, err) + assert.Contains(t, err.Error(), "consumer_activation_timeout") +} + +func TestKinesisInputEFOMaxResubscribeIntervalDefault(t *testing.T) { + pConf, err := kinesisInputSpec().ParseYAML(` +streams: [ foo ] +enhanced_fan_out: + enabled: true + consumer_name: my-app +`, nil) + require.NoError(t, err) + + conf, err := kinesisInputConfigFromParsed(pConf) + require.NoError(t, err) + assert.Equal(t, 30*time.Second, conf.EFOMaxResubscribeInterval) +} + +func TestKinesisInputEFOMaxResubscribeIntervalCustom(t *testing.T) { + pConf, err := kinesisInputSpec().ParseYAML(` +streams: [ foo ] +enhanced_fan_out: + enabled: true + consumer_name: my-app + max_resubscribe_interval: 90s +`, nil) + require.NoError(t, err) + + conf, err := kinesisInputConfigFromParsed(pConf) + require.NoError(t, err) + assert.Equal(t, 90*time.Second, conf.EFOMaxResubscribeInterval) +} + +func TestKinesisInputEFOMaxResubscribeIntervalTooLowFails(t *testing.T) { + pConf, err := kinesisInputSpec().ParseYAML(` +streams: [ foo ] +enhanced_fan_out: + enabled: true + consumer_name: my-app + max_resubscribe_interval: 500ms +`, nil) + require.NoError(t, err) + + _, err = kinesisInputConfigFromParsed(pConf) + require.Error(t, err) + assert.Contains(t, err.Error(), "max_resubscribe_interval") +} + +func TestKinesisInputEFOConsumerNameLintRule(t *testing.T) { + tests := []struct { + name string + conf string + lintPresent bool + }{ + { + name: "efo disabled", + conf: ` +aws_kinesis: + streams: [ foo ] +`, + }, + { + name: "efo enabled with consumer name", + conf: ` +aws_kinesis: + streams: [ foo ] + enhanced_fan_out: + enabled: true + consumer_name: my-app +`, + }, + { + name: "efo enabled without consumer name", + conf: ` +aws_kinesis: + streams: [ foo ] + enhanced_fan_out: + enabled: true +`, + lintPresent: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + linter := service.NewEnvironment().NewComponentConfigLinter() + + lints, err := linter.LintInputYAML([]byte(test.conf)) + require.NoError(t, err) + if test.lintPresent { + require.Len(t, lints, 1) + assert.Contains(t, lints[0].Error(), "consumer_name is required when enabled is true") + } else { + assert.Empty(t, lints) + } + }) + } +} + +func TestKinesisInputEFORequiresConsumerName(t *testing.T) { + pConf, err := kinesisInputSpec().ParseYAML(` +streams: [ foo ] +enhanced_fan_out: + enabled: true +`, nil) + require.NoError(t, err) + + _, err = kinesisInputConfigFromParsed(pConf) + require.Error(t, err) + assert.Contains(t, err.Error(), "consumer_name") +} + +func TestShardFetchWaitBound(t *testing.T) { + tests := []struct { + name string + commitPeriod time.Duration + batchPeriod time.Duration + want time.Duration + }{ + { + name: "default commit period no batch period", + commitPeriod: 5 * time.Second, + want: time.Second, + }, + { + name: "short commit period halves the bound", + commitPeriod: time.Second, + want: 500 * time.Millisecond, + }, + { + name: "batch period tightens the bound", + commitPeriod: 5 * time.Second, + batchPeriod: 100 * time.Millisecond, + want: 50 * time.Millisecond, + }, + { + name: "batch period below the floor is clamped", + commitPeriod: 5 * time.Second, + batchPeriod: time.Millisecond, + want: 5 * time.Millisecond, + }, + { + name: "zero batch period is ignored", + commitPeriod: 5 * time.Second, + batchPeriod: 0, + want: time.Second, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assert.Equal(t, test.want, shardFetchWaitBound(test.commitPeriod, test.batchPeriod)) + }) + } +} diff --git a/internal/impl/aws/kinesis/integration_test.go b/internal/impl/aws/kinesis/integration_test.go index 9ffeb3974d..765e51a1b9 100644 --- a/internal/impl/aws/kinesis/integration_test.go +++ b/internal/impl/aws/kinesis/integration_test.go @@ -94,8 +94,8 @@ func createKinesisShards(ctx context.Context, t testing.TB, awsPort, id string, return shards, nil } -func kinesisIntegrationSuite(t *testing.T, lsPort string) { - template := ` +func kinesisIntegrationTemplate(inputExtra string) string { + return ` output: aws_kinesis: endpoint: http://localhost:$PORT @@ -115,6 +115,7 @@ input: endpoint: http://localhost:$PORT streams: [ stream-$ID$VAR1 ] checkpoint_limit: $VAR2 +` + inputExtra + ` dynamodb: table: stream-$ID create: true @@ -125,6 +126,10 @@ input: secret: xxxxx token: xxxxx ` +} + +func kinesisIntegrationSuite(t *testing.T, lsPort string) { + template := kinesisIntegrationTemplate("") suite := integration.StreamTests( integration.StreamTestOpenClose(), @@ -189,3 +194,80 @@ input: ) }) } + +func TestIntegrationKinesisPollPeriod(t *testing.T) { + integration.CheckSkip(t) + + servicePort := awstest.GetLocalStack(t) + + template := kinesisIntegrationTemplate(` + poll_period: 20ms +`) + + suite := integration.StreamTests( + integration.StreamTestOpenClose(), + integration.StreamTestStreamSequential(50), + ) + suite.Run( + t, template, + integration.StreamTestOptPreTest(func(t testing.TB, ctx context.Context, vars *integration.StreamTestConfigVars) { + _, err := createKinesisShards(ctx, t, servicePort, vars.ID, 2) + require.NoError(t, err) + }), + integration.StreamTestOptPort(servicePort), + integration.StreamTestOptAllowDupes(), + integration.StreamTestOptVarSet("VAR1", ""), + integration.StreamTestOptVarSet("VAR2", "10"), + ) +} + +func TestIntegrationKinesisEFO(t *testing.T) { + integration.CheckSkip(t) + + servicePort := awstest.GetLocalStack(t) + + template := kinesisIntegrationTemplate(` + enhanced_fan_out: + enabled: true + consumer_name: rpcn-it-consumer +`) + + suite := integration.StreamTests( + integration.StreamTestOpenClose(), + integration.StreamTestSendBatch(10), + integration.StreamTestStreamSequential(200), + integration.StreamTestStreamParallel(200), + integration.StreamTestStreamParallelLossy(200), + integration.StreamTestStreamParallelLossyThroughReconnect(200), + ) + + t.Run("with balanced shards", func(t *testing.T) { + suite.Run( + t, template, + integration.StreamTestOptPreTest(func(t testing.TB, ctx context.Context, vars *integration.StreamTestConfigVars) { + _, err := createKinesisShards(ctx, t, servicePort, vars.ID, 2) + require.NoError(t, err) + }), + integration.StreamTestOptPort(servicePort), + integration.StreamTestOptAllowDupes(), + integration.StreamTestOptVarSet("VAR1", ""), + integration.StreamTestOptVarSet("VAR2", "10"), + ) + }) + + t.Run("single shard checkpointing", func(t *testing.T) { + integration.StreamTests( + integration.StreamTestCheckpointCapture(), + ).Run( + t, template, + integration.StreamTestOptPreTest(func(t testing.TB, ctx context.Context, vars *integration.StreamTestConfigVars) { + shards, err := createKinesisShards(ctx, t, servicePort, vars.ID, 1) + require.NoError(t, err) + vars.General["VAR1"] = ":" + shards[0] + }), + integration.StreamTestOptPort(servicePort), + integration.StreamTestOptAllowDupes(), + integration.StreamTestOptVarSet("VAR2", "10"), + ) + }) +}