kinesis

package
v1.6.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Sep 14, 2026 License: MIT Imports: 45 Imported by: 0

README

Kinesis

Parity grade: A · SDK aws-sdk-go-v2/service/kinesis@v1.53.0 · last audited 2026-08-19 (a4d4c728b)

Coverage

Metric Value
PARITY entries audited 39 (39 ok)
Feature families 9 (9 ok)
Known gaps 24
Deferred items 1
Resource leaks clean
Known gaps
  • UPDATE 2026-09-11 (gopherstack-s781r): the entry below (previously claiming channels 'never deliver records') is now PARTIALLY resolved. This pass re-fetched the streams dev-guide -- unlike the prior pass, WebFetch against docs.aws.amazon.com/streams/latest/dev/data-delivery-s3-key-template.html and data-delivery-s3-about.html returned full usable content this time, including the documented default OutputKeyTemplate string and its variable table. See the dated 'Channel record delivery' note below for what is now real. S3TablesDestinationConfiguration (Iceberg on Amazon S3 Tables) delivery remains UNMODELED: gopherstack has no services/s3tables data-file/manifest write path (services/s3tables only manages table-bucket/namespace/table metadata -- confirmed by grepping for PutObject/DataFile/Manifest-shaped methods there, none exist), so there is no honest way to write Parquet-in-Iceberg data even with the destination fully documented. A channel with only S3TablesDestinationConfiguration set still accepts records into its buffer's main-record list, which is now silently never flushed (see the next gap entry) -- effectively an ongoing gap for that destination type specifically, distinct from the DataFreshnessInSeconds-based flush that IS wired for S3DestinationConfiguration.
  • deliverPutToChannels (channel_delivery.go) only matches channels with a non-nil S3DestinationConfiguration -- a channel whose only destination is S3TablesDestinationConfiguration is filtered out before ever reaching appendToChannelBuffer, so it never buffers or accumulates records at all (deliberately: buffering records with no delivery path would be worse than not buffering them -- they'd sit in memory pretending to be 'in flight' for a destination this backend cannot honestly write to). Records put to such a channel's source stream are simply never observed by the delivery layer, same end effect as before this pass but now for a narrower, correctly-scoped reason. (gopherstack-s781r follow-up)
  • S3 output key template VALIDATION (docs.aws.amazon.com/streams/latest/dev/data-delivery-s3-key-template.html's 'Template rules': 1024-char object-key cap after the ~38-char unique suffix, no path traversal, no consecutive slashes, single extension placeholder only at the end, restricted literal charset) is not enforced at CreateChannel/UpdateChannel time -- an OutputKeyTemplate violating these rules is accepted, and buildChannelObjectKey (channel_delivery.go) will still expand it as best-effort at flush time rather than rejecting it upfront the way real AWS's CreateChannel validator would. Key EXPANSION itself (variable substitution, default template, extension derivation) is real and tested. (gopherstack-s781r follow-up)
  • The unique suffix S3 delivery documents as always appended to every object key ('Amazon Kinesis Data Streams automatically appends a unique suffix to every object key') has no documented insertion point or format (length/charset) in the fetched docs. buildChannelObjectKey inserts a 12-character slice of a UUIDv4 immediately before the file extension (or at the end, with no extension), mirroring the position Firehose's own buildS3Key (services/firehose/delivery_s3.go) uses for its uniqueness token -- disclosed as an inference, not a verified AWS behavior.
  • The exact byte-level layout of a delivered S3 object's body is not documented beyond data-delivery.html's 'Records are delivered in their original source format with no transformation applied' (general purpose S3 destinations only -- GSR_JSON/Iceberg conversion is documented separately for streaming tables, which this backend does not implement). writeChannelObject (channel_delivery.go) therefore concatenates each buffered record's raw bytes with NO delimiter inserted between records, the literal reading of 'no transformation applied'. Not verified against a captured real AWS object; disclosed as the most literal reading of the documented behavior.
  • The dead-letter queue's exact object schema is documented only at the field-list level (data-delivery-s3-about.html / the 'What's New' announcement: 'stream ARN, shard ID, sequence number, and error context'), with no documented JSON key names or file layout. writeChannelDeadLetterQueue (channel_delivery.go) writes newline-delimited JSON with keys streamARN/shardID/sequenceNumber/errorMessage, one line per failed record -- disclosed as an inference, not a verified wire format. The default dead-letter prefix used when DeadLetterQueueS3Configuration is unset (defaultChannelErrorPrefix = "kinesis-channel-errors/") is likewise inferred: AWS documents only the behavior ('defaults to the destination bucket with an error prefix'), not the literal prefix string.
  • DataFreshnessInSeconds is confirmed (both by the pinned SDK's S3DestinationConfiguration/S3StorageConfiguration Go types and by data-delivery-s3-about.html) to be the ONLY documented buffering control for channel S3 delivery -- there is no separate size-based BufferingHints field on S3DestinationConfiguration (unlike Firehose's S3DestinationDescription.BufferingHints.SizeInMBs). Flush is therefore purely interval-based here, which is a verified-correct simplification, not a disclosed gap by itself; noted so a future pass doesn't assume a missing size trigger is a bug.
  • Kinesis has NO injectable clock anywhere in this backend (checked: no Clock interface, no nowFn/timeSource field on InMemoryBackend; the existing janitor.go retention sweeper also uses time.Now() directly on a real time.Ticker). runChannelFlusher (channel_delivery.go) therefore polls a real 1-second time.Ticker for DataFreshnessInSeconds-elapsed channels, mirroring Firehose's own intervalFlusher (services/firehose/flush.go), which has the same real-ticker, no-injectable-clock design. FlushChannel/FlushAllChannels are exported so tests and DeleteChannel/Handler.Shutdown can force an immediate flush without waiting on or sleeping past the ticker.
  • Buffered-but-unflushed channel records are NOT persisted across a Snapshot/Restore cycle -- channelBuffers is in-memory-only state on InMemoryBackend, not part of backendSnapshot. Handler.Shutdown (mirroring Firehose's) best-effort flushes every channel via FlushAllChannels before the process exits, and DeleteChannel/DeleteStream flush their affected channel(s) before removing state, which covers graceful shutdown and explicit deletion; an ungraceful process exit (crash, SIGKILL) between an accepted PutRecord and the next flush still loses that channel's currently-buffered records on restart. Disclosed rather than silently accepted; no snapshot_inventory.json changes were needed since no new persisted field was added.
  • Channel ARN format (arn:{partition}:kinesis:{region}:{accountID}:channel/{channelName}) is inferred by following the same '{service}/{resource-name}' convention AWS uses for every other Kinesis resource (stream/{name}, stream/{name}/consumer/{name}) -- the pinned SDK's doc comments give no ARN format for channels at all (unlike streams/consumers, documented in the IAM access-control guide, which itself predates the channels feature and was re-fetched this pass with no channel-ARN mention added). Not verified against a real AWS response; disclosed rather than asserted as confirmed.
  • CreateChannel/DeleteChannel/DescribeChannel/ListChannels/UpdateChannel's documented 5 TPS-per-account call-limit LimitExceededException is not modeled. The service already has a couple of injectable-clock throttle precedents elsewhere in the codebase (e.g. services/polly's per-engine sliding-window throttle), but wiring an equivalent per-op rate model into this already-large file was judged disproportionate to this pass's ask; not fabricated. LimitExceededException remains reachable through this service's other existing rate-limited paths (tag limits, consumer-registration cap) -- it is only the channel-specific 5 TPS window that is unmodeled.
  • ChannelDescription/ChannelSummary's S3TablesConfiguration.PartitionSpec is modeled and round-trips (ChannelPartitionSpec/ChannelPartitionField), but this backend performs no actual Iceberg partitioning -- there is no partitioning behavior to verify the accepted spec against, only storage/echo.
  • KMSAccessDeniedException (types.KMSAccessDeniedException) is a real modeled StartStreamEncryption/StopStreamEncryption error but has no trigger path: it requires evaluating a KMS key policy/grant against a calling principal, and gopherstack has no IAM policy evaluation engine anywhere (not just in kinesis) to produce an access-denied decision from. The sentinel (ErrKMSAccessDenied) and its InvalidArgumentException-style wire mapping (KMSAccessDeniedException, 400) are defined for wire-shape completeness, matching the real error type string exactly, but nothing in the backend can ever return it. Fabricating a fake denial rule (e.g. 'deny if KeyId contains X') would itself be a stub, so this stays an honest gap rather than a fake implementation. (bd: gopherstack-ud2)
  • RESOLVED 2026-09-11 (gopherstack-s0ju item 2): the entry previously here claimed UpdateStreamMode's PROVISIONED -> ON_DEMAND transition 'approximates AWS's real throughput-history-based scaling with a fixed floor' -- that floor-reshard-at-transition-time behavior has been removed entirely (it directly contradicted the doc: real AWS keeps the pre-transition shard count on this exact transition, with no immediate reshard). What remains approximated, honestly, in the new reactive maybeAutoScaleOnDemand (ondemand_scaling.go, see UpdateStreamMode's own ops: note for the full citation): a 60-second write-rate window stands in for AWS's real 30-day peak-throughput history; a stream-wide shard-count doubling stands in for AWS's per-shard hot-shard split; and '500 KB/s per shard' is read as 500 KiB/s, an inferred unit. (bd: gopherstack-ud2)
  • RESOLVED 2026-09-11 (gopherstack-s0ju item 3): the entry previously here described AT_TRIM_HORIZON's ListShards ShardFilter clamp (trimHorizon, shards.go) as also standing in for per-record retention filtering more broadly -- that conflation was itself a gap. trimHorizon's clamp-to-oldest-shard-StartedAt behavior is correct and unchanged for its actual, narrow use (ListShards' shard-existence/lineage filtering, where an empty result for a freshly created stream would be wrong). It is no longer used for GetShardIterator/SubscribeToShard's record-level TRIM_HORIZON/AT_TIMESTAMP, which now call the new, unclamped retentionCutoff directly (see GetShardIterator's own ops: note) and therefore honor true per-record ApproximateArrivalTimestamp-vs-retention semantics, including immediately after a DecreaseStreamRetentionPeriod and before the janitor's next sweep. No remaining approximation gap for GetShardIterator/SubscribeToShard's retention handling specifically.
  • CORRECTED this pass: the previous gap entry claiming resource policies (PutResourcePolicy/GetResourcePolicy/DeleteResourcePolicy) are lost across a persistence restart was stale/incorrect. persistence.go's backendSnapshot already has a ResourcePolicies field wired into both Snapshot (line ~60) and Restore (line ~119), and TestInMemoryBackend_FullStateSnapshotRestoreRoundTrip already exercises PutResourcePolicy through an actual snapshot/restore cycle and passes. No code change needed; this was a documentation-only correction (carried forward unchanged from the prior ledger).
  • CORRECTED this pass: the deferred entry below claiming Lambda event-source-mapping trigger wiring 'lives in cli.go per task constraints; not touched' was stale -- cli.go's wireKinesisLambda (called at cli.go:2657) already wires services/kinesis to services/lambda's event-source poller via kinesisReaderAdapter, and this has been true since before this pass. Moved out of deferred; documentation-only correction, no code changed for this item.
  • AccessDeniedException is declared in UpdateMaxRecordSize's and UpdateStreamWarmThroughput's error switches (deserializers.go's awsAwsjson11_deserializeOpErrorUpdateMaxRecordSize / ...UpdateStreamWarmThroughput both list it) but has no trigger path, for the same reason as KMSAccessDeniedException above: no IAM policy evaluation engine anywhere in gopherstack to produce an access-denied decision from. Not fabricated a fake rule for it; stays an honest gap. (gopherstack-nbg8)
  • ResourceInUseException is declared for both UpdateMaxRecordSize and UpdateStreamWarmThroughput (real AWS returns it when the target stream isn't ACTIVE, e.g. mid-UPDATING from a concurrent operation) but is unreachable here: every stream in this backend is ACTIVE immediately on creation and stays so (no transient CREATING/UPDATING/DELETING window is modeled anywhere in kinesis, not just for these two ops), so there is never a state where the check could fire honestly. (gopherstack-nbg8)
  • UpdateStreamWarmThroughput applies the requested WarmThroughputMiBps synchronously; real AWS documents this as asynchronous (stream goes UPDATING, then back to ACTIVE, 'could take a few minutes to complete' for a large stream). This backend has no transient-state model for any stream-level operation (see the ResourceInUseException gap above), so WarmThroughputObject.CurrentMiBps and TargetMiBps always match immediately on read instead of Current lagging Target during a scale-up window. (gopherstack-nbg8)
  • 2026-08-14 (gopherstack-enpq): mechanical struct-field diff (cmd/structfielddiff) against aws-sdk-go-v2/service/kinesis@v1.46.4, a different method than the op-by-op audits above (gopherstack-nbg8/r80d, 2026-08-13). Only 4 of 39 ops had a real candidate field miss after excluding ResultMetadata/StreamId (both known noise -- StreamId is 'Not Implemented. Reserved for future use.' on every op that has it, same class as sqs's already-known-noise fields), a low false-positive rate that independently re-confirms this service's very recent A-grade sweep by a different method than the reads that earned it. 2 real bugs FOUND AND FIXED: DeleteStream's missing EnforceConsumerDeletion gate (see DeleteStream ops entry) and GetRecords' missing ChildShards plus the NextShardIterator empty-string-vs-null bug it uncovered alongside it (see GetRecords ops entry). PutRecordInput.SequenceNumberForOrdering confirmed a non-issue (see PutRecord ops entry). ListStreamsOutput.StreamSummaries disclosed below, not fixed.
  • CORRECTED (gopherstack-enpq, 2026-08-22): the 2026-08-14 entry above's 'only 4 of 39 ops had a real candidate field miss' / 'low false-positive rate ... independently re-confirms' claim was overstated -- re-running structfielddiff and hand-verifying every op's request-side identifier contract against its own api_op_*.go doc comment (not just its field list) found 9 more real bugs the prior pass missed: AddTagsToStream/RemoveTagsFromStream/ListTagsForStream/IncreaseStreamRetentionPeriod/DecreaseStreamRetentionPeriod/ListShards/EnableEnhancedMonitoring/DisableEnhancedMonitoring/UpdateShardCount all had no Go field for StreamARN despite each op's own doc comment requiring support for it ('you must use either the StreamARN or the StreamName parameter... recommended that you use the StreamARN input parameter') -- plus RegisterStreamConsumer.Tags (structurally absent) and DescribeStreamSummary.MaxRecordSizeInKiB/WarmThroughput (tracked internally, never surfaced). All 11 fixed this pass; see the dated section below and each op's own ops: entry.
  • DISCLOSED, not fixed (2026-08-19 wrapper-key/nested-shape sweep, Layer 3 -- never-emitted optional members, explicitly out of scope as a hunt per that sweep's charter): StreamDescriptionSummary is missing three optional real members it could populate from backend state already tracked on Stream -- MaxRecordSizeInKiB (Stream.MaxRecordSizeBytes / bytesPerKiB), StreamId (n/a -- real AWS documents this as 'Not Implemented. Reserved for future use.', same known-noise class as other StreamId fields), and WarmThroughput (Stream.WarmThroughputMiBps, same WarmThroughputObject shape UpdateStreamWarmThroughput already emits correctly). UpdateShardCountOutput.StreamARN (optional) is also never emitted -- UpdateShardCountOutput (models.go) has no StreamARN field at the backend-output level, so this needs backend plumbing, not a one-line wire fix. Record.EncryptionType (optional, both GetRecordsOutput.Records and SubscribeToShardEvent.Records) is never emitted -- this backend does track Stream.EncryptionType but doesn't thread it onto individual jsonRecord entries. None of these are wrong keys or wrong types; they are members with no case reached at all because nothing writes them. (bd: gopherstack-ud2)
  • DISCLOSED, not fixed (gopherstack-enpq): ListStreamsOutput.StreamSummaries ([]types.StreamSummary -- ARN/name/status/creation-timestamp/mode per stream, optional not required) is not populated; only the required StreamNames is. Real AWS's newer SDKs/console traffic favor StreamSummaries over the legacy StreamNames-only shape, so a client reading only StreamSummaries would see an empty list even though StreamNames (the field the real validator actually requires) is correct. Not fixed this pass: the backend's ListStreams pagination is built entirely around a sorted []string of names (streams.go), and building StreamSummaries correctly means carrying the full *Stream (or at least ARN/Status/CreatedAt/StreamMode) through that same pagination window rather than bolting a lookup on afterward -- a real reshape of ListStreamsOutput/the backend method signature, not a one-line add, so it was left disclosed rather than rushed. (bd: gopherstack-ud2)
Deferred
  • RESOLVED 2026-09-11 (gopherstack-s0ju item 4): this entry previously read 'Enhanced fan-out SubscribeToShard real streaming cadence / HTTP2 push semantics beyond the polling emulation already in place.' The emulator still polls internally (no injectable I/O push mechanism) rather than truly pushing over HTTP/2, but the previously-deferred cadence divergence is fixed: the stream now stays open for the documented 5-minute window and sends periodic heartbeats instead of self-closing after 3 empty polls (~600ms) -- see SubscribeToShard's own ops: note for the full citation and remaining disclosed approximations (heartbeat interval not documented exactly by AWS; ChildShards still not populated on any SubscribeToShardEvent, data or heartbeat -- see the existing ChildShards gap above).

More

Documentation

Index

Constants

View Source
const (

	// StreamModeProvisioned is the PROVISIONED stream mode.
	StreamModeProvisioned = "PROVISIONED"
	// StreamModeOnDemand is the ON_DEMAND stream mode.
	StreamModeOnDemand = "ON_DEMAND"
)

Variables

View Source
var (
	ErrStreamNotFound      = awserr.New("ResourceNotFoundException", awserr.ErrNotFound)
	ErrStreamAlreadyExists = awserr.New("ResourceInUseException", awserr.ErrAlreadyExists)
	// ErrStreamHasConsumers is returned by DeleteStream when the stream has
	// registered enhanced fan-out consumers and EnforceConsumerDeletion is
	// unset or false (real DeleteStreamInput.EnforceConsumerDeletion doc
	// comment: "the call to DeleteStream fails with a ResourceInUseException").
	ErrStreamHasConsumers     = awserr.New("ResourceInUseException", awserr.ErrConflict)
	ErrInvalidArgument        = awserr.New("InvalidArgumentException", awserr.ErrInvalidParameter)
	ErrUnknownAction          = errors.New("UnknownOperationException")
	ErrShardIteratorExpired   = errors.New("ExpiredIteratorException")
	ErrConsumerNotFound       = awserr.New("ResourceNotFoundException", awserr.ErrNotFound)
	ErrConsumerAlreadyExists  = awserr.New("ResourceInUseException", awserr.ErrAlreadyExists)
	ErrResourcePolicyNotFound = awserr.New("ResourceNotFoundException", awserr.ErrNotFound)
	// ErrChannelNotFound is returned by DescribeChannel/UpdateChannel/
	// DeleteChannel when ChannelARN does not match any channel, and by
	// ListTagsForResource/TagResource/UntagResource for an unrecognized
	// channel ARN.
	ErrChannelNotFound = awserr.New("ResourceNotFoundException", awserr.ErrNotFound)
	// ErrChannelAlreadyExists is returned by CreateChannel when ChannelName
	// collides with an existing channel: "The name is unique within your
	// Amazon Web Services account and Amazon Web Services Region"
	// (api_op_CreateChannel.go doc comment on ChannelName). ResourceInUseException
	// is one of CreateChannel's declared exceptions (deserializers.go
	// deserializeOpErrorCreateChannel).
	ErrChannelAlreadyExists          = awserr.New("ResourceInUseException", awserr.ErrAlreadyExists)
	ErrProvisionedThroughputExceeded = awserr.New(
		"ProvisionedThroughputExceededException",
		errRateExceeded,
	)
	ErrTagLimitExceeded = awserr.New("LimitExceededException", awserr.ErrInvalidParameter)
	ErrLimitExceeded    = awserr.New("LimitExceededException", awserr.ErrInvalidParameter)

	// ErrKMSNotFound indicates the KMS key referenced by StartStreamEncryption's
	// KeyId does not exist. Only reachable when a KMSKeyValidator is wired via
	// WithKMSValidator (see stream_encryption.go); with no validator wired,
	// KeyId is format-checked only, matching pre-existing behavior for
	// deployments that don't wire cross-service KMS validation.
	ErrKMSNotFound = errors.New("KMSNotFoundException")
	// ErrKMSDisabled indicates the KMS key exists but is disabled or pending
	// deletion/import, matching the real KMSDisabledException/
	// KMSInvalidStateException split (see stream_encryption.go for which
	// applies to which key state).
	ErrKMSDisabled = errors.New("KMSDisabledException")
	// ErrKMSInvalidState indicates the KMS key is in a state (e.g. pending
	// deletion, pending import) that doesn't allow use for encryption.
	ErrKMSInvalidState = errors.New("KMSInvalidStateException")
	// ErrKMSAccessDenied is modeled per the real aws-sdk-go-v2/service/kinesis
	// error set for StartStreamEncryption/StopStreamEncryption/
	// UpdateMaxRecordSize (types.KMSAccessDeniedException), but gopherstack has
	// no IAM/key-policy evaluation engine to ever produce it -- defined for
	// wire-shape completeness and left with no trigger path (see PARITY.md gaps).
	ErrKMSAccessDenied = errors.New("KMSAccessDeniedException")
)

Sentinel errors for Kinesis operations.

View Source
var ErrNilAppContext = errors.New("kinesis: AppContext must not be nil")

ErrNilAppContext is returned by Init when the AppContext is nil.

View Source
var ErrShardCountScaling = errors.New(
	"UpdateShardCount cannot scale by more than double or less than half " +
		"of the current shard count within a single call",
)

ErrShardCountScaling indicates that an UpdateShardCount target fell outside the AWS per-call scaling window: within a single call the target shard count may not be more than double or less than half of the current open shard count. AWS surfaces this as ValidationException. The sentinel's message is the error text returned to the client.

View Source
var ErrValidation = errors.New("kinesis: validation error")

ErrValidation is the sentinel error for Kinesis input validation failures.

Functions

func ContextAndNameFromStreamARN

func ContextAndNameFromStreamARN(ctx context.Context, streamARN string) (context.Context, string)

ContextAndNameFromStreamARN parses a Kinesis stream ARN (arn:aws:kinesis:{region}:{account}:stream/{name}) and returns ctx with the ARN's region attached alongside the plain stream name, for callers outside this package that only hold the ARN -- cli.go's wireKinesisLambda (gopherstack-qowd), so Kinesis-to-Lambda event source mappings resolve the stream's actual region instead of always the account default.

Types

type Channel

type Channel struct {
	ChannelCreationTimestamp         time.Time                   `json:"channelCreationTimestamp"`
	EncryptionConfiguration          *ChannelEncryptionConfig    `json:"encryptionConfiguration,omitempty"`
	S3DestinationConfiguration       *ChannelS3Destination       `json:"s3DestinationConfiguration,omitempty"`
	S3TablesDestinationConfiguration *ChannelS3TablesDestination `json:"s3TablesDestinationConfiguration,omitempty"`
	Tags                             map[string]string           `json:"tags,omitempty"`
	ChannelID                        string                      `json:"channelID"`
	ChannelARN                       string                      `json:"channelARN"`
	ChannelName                      string                      `json:"channelName"`
	ChannelStatus                    string                      `json:"channelStatus"`
	ServiceExecutionRoleARN          string                      `json:"serviceExecutionRoleARN"`
	Region                           string                      `json:"region,omitempty"`
	StreamConfigurationList          []ChannelStreamConfig       `json:"streamConfigurationList"`
	LoggingConfiguration             ChannelCloudWatchLogsConfig `json:"loggingConfiguration"`
}

Channel represents an in-memory Kinesis Data Streams channel -- a CreateChannel-provisioned delivery pipe from a stream to either a general purpose S3 bucket or a streaming table (Apache Iceberg / Amazon S3 Tables) destination (types.ChannelDescription, kinesis@v1.53.0 types/types.go:9-79). Records put to the source stream are NOT delivered to either destination by this backend -- see PARITY.md.

type ChannelCloudWatchLogsConfig

type ChannelCloudWatchLogsConfig struct {
	LogGroupName  string `json:"logGroupName,omitempty"`
	LogStreamName string `json:"logStreamName,omitempty"`
	Enabled       bool   `json:"enabled"`
}

ChannelCloudWatchLogsConfig mirrors types.CloudWatchLogs / types.CloudWatchLogsUpdateInput (kinesis@v1.53.0 types/types.go:256-289).

type ChannelDeadLetterQueueS3Config

type ChannelDeadLetterQueueS3Config struct {
	BucketARN           string `json:"bucketARN"`
	ExpectedBucketOwner string `json:"expectedBucketOwner,omitempty"`
	ErrorOutputPrefix   string `json:"errorOutputPrefix,omitempty"`
}

ChannelDeadLetterQueueS3Config mirrors types.DeadLetterQueueS3Configuration (kinesis@v1.53.0 types/types.go:364-381).

type ChannelEncryptionConfig

type ChannelEncryptionConfig struct {
	EncryptionType string `json:"encryptionType"`
	KeyID          string `json:"keyID"`
}

ChannelEncryptionConfig mirrors types.ChannelEncryptionConfiguration (kinesis@v1.53.0 types/types.go:83-97).

type ChannelPartitionField

type ChannelPartitionField struct {
	SourceName string `json:"sourceName"`
	Transform  string `json:"transform"`
}

ChannelPartitionField mirrors types.PartitionField (kinesis@v1.53.0 types/types.go:479-493).

type ChannelPartitionSpec

type ChannelPartitionSpec struct {
	PartitionFields []ChannelPartitionField `json:"partitionFields"`
}

ChannelPartitionSpec mirrors types.PartitionSpec (kinesis@v1.53.0 types/types.go:495-504).

type ChannelRecordConfig

type ChannelRecordConfig struct {
	RecordFormatType string `json:"recordFormatType"`
	GSRSchemaARN     string `json:"gsrSchemaARN,omitempty"`
}

ChannelRecordConfig mirrors types.RecordConfiguration (kinesis@v1.53.0 types/types.go:600-624).

type ChannelS3Destination

type ChannelS3Destination struct {
	DeadLetterQueueS3Configuration *ChannelDeadLetterQueueS3Config `json:"deadLetterQueueS3Configuration,omitempty"`
	StorageConfiguration           ChannelS3StorageConfig          `json:"storageConfiguration"`
	DataFreshnessInSeconds         int                             `json:"dataFreshnessInSeconds"`
}

ChannelS3Destination mirrors the merged shape of types.S3DestinationConfiguration (CreateChannelInput) and types.S3DestinationDescription (ChannelDescription) -- both carry the same members, only required-ness differs between the two (kinesis@v1.53.0 types/types.go:626-663).

type ChannelS3StorageConfig

type ChannelS3StorageConfig struct {
	BucketARN           string `json:"bucketARN"`
	CompressionType     string `json:"compressionType"`
	ExpectedBucketOwner string `json:"expectedBucketOwner,omitempty"`
	OutputKeyTemplate   string `json:"outputKeyTemplate,omitempty"`
	StorageClass        string `json:"storageClass,omitempty"`
}

ChannelS3StorageConfig mirrors types.S3StorageConfiguration (kinesis@v1.53.0 types/types.go:678-714).

type ChannelS3TablesConfig

type ChannelS3TablesConfig struct {
	PartitionSpec   *ChannelPartitionSpec `json:"partitionSpec,omitempty"`
	TableBucketARN  string                `json:"tableBucketARN"`
	Namespace       string                `json:"namespace"`
	TableName       string                `json:"tableName"`
	CompressionType string                `json:"compressionType"`
}

ChannelS3TablesConfig mirrors types.S3TablesConfiguration (kinesis@v1.53.0 types/types.go:718-746).

type ChannelS3TablesDestination

type ChannelS3TablesDestination struct {
	DeadLetterQueueS3Configuration *ChannelDeadLetterQueueS3Config `json:"deadLetterQueueS3Configuration,omitempty"`
	S3TablesConfigurationList      []ChannelS3TablesConfig         `json:"s3TablesConfigurationList"`
	DataFreshnessInSeconds         int                             `json:"dataFreshnessInSeconds"`
}

ChannelS3TablesDestination mirrors types.S3TablesDestinationConfiguration / types.S3TablesDestinationDescription (kinesis@v1.53.0 types/types.go:762-802).

type ChannelS3Writer

type ChannelS3Writer interface {
	PutObject(ctx context.Context, input *sdk_s3.PutObjectInput) (*sdk_s3.PutObjectOutput, error)
}

ChannelS3Writer is the subset of S3 operations that channel record delivery needs to write an object to a general purpose S3 destination. Mirrors firehose.S3Storer (services/firehose/interfaces.go).

type ChannelStreamConfig

type ChannelStreamConfig struct {
	StreamCreationTimestamp time.Time           `json:"streamCreationTimestamp"`
	StreamARN               string              `json:"streamARN"`
	RecordConfiguration     ChannelRecordConfig `json:"recordConfiguration"`
}

ChannelStreamConfig mirrors the source-stream binding shared by types.ChannelStreamConfiguration (CreateChannelInput's StreamConfigurationList), types.ChannelStreamDescription (ChannelDescription's StreamConfigurationList), and types.ChannelStreamIdentifier (ChannelSummary's Streams) -- kinesis@v1.53.0 types/types.go:122-172.

type ChannelStreamFilter

type ChannelStreamFilter struct {
	StreamARN string
}

ChannelStreamFilter mirrors types.StreamFilter (kinesis@v1.53.0 types/types.go:1141-1153), used by ListChannels to filter by source stream.

type ChildShard added in v1.3.1

type ChildShard struct {
	ShardID           string
	HashKeyRangeStart string
	HashKeyRangeEnd   string
	ParentShards      []string
}

ChildShard describes a shard that resulted from splitting or merging the shard a GetRecords call just finished reading (aws-sdk-go-v2 types.ChildShard). Real AWS only returns this "when the end of the current shard is reached" -- i.e. exactly when NextShardIterator is empty because the shard is Closed and fully consumed.

type ConfigProvider

type ConfigProvider interface {
	GetKinesisSettings() Settings
}

ConfigProvider is a private interface to extract Kinesis configuration from the abstract AppContext Config.

type Consumer

type Consumer struct {
	ConsumerCreationTimestamp time.Time         `json:"consumerCreationTimestamp"`
	Tags                      map[string]string `json:"tags,omitempty"`
	ConsumerName              string            `json:"consumerName"`
	ConsumerARN               string            `json:"consumerARN"`
	ConsumerStatus            string            `json:"consumerStatus"`
	StreamARN                 string            `json:"streamARN"`
}

Consumer represents a registered Kinesis enhanced fan-out consumer.

type CreateChannelInput

type CreateChannelInput struct {
	EncryptionConfiguration          *ChannelEncryptionConfig
	LoggingConfiguration             *ChannelCloudWatchLogsConfig
	S3DestinationConfiguration       *ChannelS3Destination
	S3TablesDestinationConfiguration *ChannelS3TablesDestination
	Tags                             map[string]string
	ChannelName                      string
	ServiceExecutionRoleARN          string
	StreamConfigurationList          []ChannelStreamConfig
}

CreateChannelInput is the input for CreateChannel.

type CreateChannelOutput

type CreateChannelOutput struct {
	ChannelDescription Channel
}

CreateChannelOutput is the output for CreateChannel.

type CreateStreamInput

type CreateStreamInput struct {
	StreamName string
	Region     string
	AccountID  string
	StreamMode string
	ShardCount int
	// MaxRecordSizeInKiB mirrors CreateStreamInput's own field of the same
	// name (kinesis@v1.46.4 api_op_CreateStream.go:101-103), not just
	// UpdateMaxRecordSize's. Zero means "not specified" -- CreateStream keeps
	// defaultMaxRecordSizeBytes, same as an omitted request member.
	MaxRecordSizeInKiB int
	// WarmThroughputMiBps mirrors CreateStreamInput's own field of the same
	// name (kinesis@v1.46.4 api_op_CreateStream.go:119-121). Zero means "not
	// specified".
	WarmThroughputMiBps int
}

CreateStreamInput is the input for CreateStream.

type DecreaseStreamRetentionPeriodInput

type DecreaseStreamRetentionPeriodInput struct {
	StreamName           string
	RetentionPeriodHours int
}

DecreaseStreamRetentionPeriodInput is the input for DecreaseStreamRetentionPeriod.

type DeleteChannelInput

type DeleteChannelInput struct {
	ChannelARN string
}

DeleteChannelInput is the input for DeleteChannel.

type DeleteResourcePolicyInput

type DeleteResourcePolicyInput struct {
	ResourceARN string
}

DeleteResourcePolicyInput is the input for DeleteResourcePolicy.

type DeleteStreamInput

type DeleteStreamInput struct {
	StreamName string
	// EnforceConsumerDeletion mirrors the real DeleteStreamInput field: unset
	// or false with registered consumers fails the call with
	// ResourceInUseException instead of deleting the stream.
	EnforceConsumerDeletion bool
}

DeleteStreamInput is the input for DeleteStream.

type DeregisterStreamConsumerInput

type DeregisterStreamConsumerInput struct {
	StreamARN    string
	ConsumerARN  string
	ConsumerName string
}

DeregisterStreamConsumerInput is the input for DeregisterStreamConsumer.

type DescribeAccountSettingsOutput

type DescribeAccountSettingsOutput struct {
	MinimumThroughputBillingCommitment MinimumThroughputBillingCommitmentOutput
}

DescribeAccountSettingsOutput is the output for DescribeAccountSettings.

type DescribeChannelInput

type DescribeChannelInput struct {
	ChannelARN string
}

DescribeChannelInput is the input for DescribeChannel.

type DescribeChannelOutput

type DescribeChannelOutput struct {
	ChannelDescription Channel
}

DescribeChannelOutput is the output for DescribeChannel.

type DescribeStreamConsumerInput

type DescribeStreamConsumerInput struct {
	StreamARN    string
	ConsumerARN  string
	ConsumerName string
}

DescribeStreamConsumerInput is the input for DescribeStreamConsumer.

type DescribeStreamConsumerOutput

type DescribeStreamConsumerOutput struct {
	ConsumerDescription Consumer
}

DescribeStreamConsumerOutput is the output for DescribeStreamConsumer.

type DescribeStreamInput

type DescribeStreamInput struct {
	StreamName string
	// ExclusiveStartShardID resumes shard pagination after the given shard ID.
	ExclusiveStartShardID string
	// Limit caps the number of ShardDescription entries returned (AWS default
	// 100, max 10000). Zero means "use the AWS default".
	Limit int
}

DescribeStreamInput is the input for DescribeStream.

type DescribeStreamOutput

type DescribeStreamOutput struct {
	StreamCreationTimestamp time.Time
	StreamName              string
	StreamARN               string
	StreamStatus            string
	EncryptionType          string
	StreamMode              string
	KeyID                   string
	Shards                  []ShardDescription
	EnhancedMonitoring      []string
	RetentionPeriodHours    int
	// HasMoreShards indicates the shard list was truncated by Limit and more
	// shards can be fetched with a follow-up call using ExclusiveStartShardID.
	HasMoreShards bool
	// MaxRecordSizeBytes and WarmThroughputMiBps mirror the same-named Stream
	// fields (see UpdateMaxRecordSize/UpdateStreamWarmThroughput). Real
	// StreamDescriptionSummary carries both (MaxRecordSizeInKiB/WarmThroughput);
	// StreamDescription (DescribeStream) does not, so only
	// handleDescribeStreamSummary reads these.
	MaxRecordSizeBytes  int
	WarmThroughputMiBps int
}

DescribeStreamOutput is the output for DescribeStream.

type DisableEnhancedMonitoringInput

type DisableEnhancedMonitoringInput struct {
	StreamName        string
	ShardLevelMetrics []string
}

DisableEnhancedMonitoringInput is the input for DisableEnhancedMonitoring.

type DisableEnhancedMonitoringOutput

type DisableEnhancedMonitoringOutput struct {
	StreamName               string
	StreamARN                string
	CurrentShardLevelMetrics []string
	DesiredShardLevelMetrics []string
}

DisableEnhancedMonitoringOutput is the output for DisableEnhancedMonitoring.

type EnableEnhancedMonitoringInput

type EnableEnhancedMonitoringInput struct {
	StreamName        string
	ShardLevelMetrics []string
}

EnableEnhancedMonitoringInput is the input for EnableEnhancedMonitoring.

type EnableEnhancedMonitoringOutput

type EnableEnhancedMonitoringOutput struct {
	StreamName               string
	StreamARN                string
	CurrentShardLevelMetrics []string
	DesiredShardLevelMetrics []string
}

EnableEnhancedMonitoringOutput is the output for EnableEnhancedMonitoring.

type GetRecordResult

type GetRecordResult struct {
	ApproximateArrivalTimestamp time.Time
	PartitionKey                string
	SequenceNumber              string
	EncryptionType              string
	Data                        []byte
}

GetRecordResult is a single record returned by GetRecords.

type GetRecordsInput

type GetRecordsInput struct {
	ShardIterator string
	Limit         int
}

GetRecordsInput is the input for GetRecords.

type GetRecordsOutput

type GetRecordsOutput struct {
	NextShardIterator  string
	Records            []GetRecordResult
	ChildShards        []ChildShard
	MillisBehindLatest int64
}

GetRecordsOutput is the output for GetRecords.

type GetResourcePolicyInput

type GetResourcePolicyInput struct {
	ResourceARN string
}

GetResourcePolicyInput is the input for GetResourcePolicy.

type GetResourcePolicyOutput

type GetResourcePolicyOutput struct {
	Policy string
}

GetResourcePolicyOutput is the output for GetResourcePolicy.

type GetShardIteratorInput

type GetShardIteratorInput struct {
	Timestamp              *time.Time
	StreamName             string
	ShardID                string
	ShardIteratorType      string
	StartingSequenceNumber string
}

GetShardIteratorInput is the input for GetShardIterator. Timestamp is a pointer so a genuinely omitted value (nil) can be distinguished from an explicit epoch-zero timestamp; required (non-nil) when ShardIteratorType is AT_TIMESTAMP.

type GetShardIteratorOutput

type GetShardIteratorOutput struct {
	ShardIterator string
}

GetShardIteratorOutput is the output for GetShardIterator.

type Handler

type Handler struct {
	Backend StorageBackend

	DefaultRegion string
	AccountID     string
	// contains filtered or unexported fields
}

Handler is the Echo HTTP handler for Kinesis operations.

func NewHandler

func NewHandler(backend StorageBackend) *Handler

NewHandler creates a new Kinesis Handler.

func (*Handler) ChaosOperations

func (h *Handler) ChaosOperations() []string

ChaosOperations returns all operations that can be fault-injected.

func (*Handler) ChaosRegions

func (h *Handler) ChaosRegions() []string

ChaosRegions returns all regions this Kinesis instance handles.

func (*Handler) ChaosServiceName

func (h *Handler) ChaosServiceName() string

ChaosServiceName returns the lowercase AWS service name for fault rule matching.

func (*Handler) ExecuteFISAction

func (h *Handler) ExecuteFISAction(ctx context.Context, action service.FISActionExecution) error

ExecuteFISAction executes a FIS action against resolved Kinesis targets.

func (*Handler) ExtractOperation

func (h *Handler) ExtractOperation(c *echo.Context) string

ExtractOperation extracts the Kinesis action from the X-Amz-Target header.

func (*Handler) ExtractResource

func (h *Handler) ExtractResource(c *echo.Context) string

ExtractResource extracts the stream name from the JSON request body.

func (*Handler) FISActions

func (h *Handler) FISActions() []service.FISActionDefinition

FISActions returns the FIS action definitions that the Kinesis service supports.

func (*Handler) GetSupportedOperations

func (h *Handler) GetSupportedOperations() []string

GetSupportedOperations returns the list of supported Kinesis operations.

func (*Handler) Handler

func (h *Handler) Handler() echo.HandlerFunc

Handler returns the Echo handler function for Kinesis operations.

func (*Handler) MatchPriority

func (h *Handler) MatchPriority() int

MatchPriority returns the routing priority for the Kinesis handler.

func (*Handler) Name

func (h *Handler) Name() string

Name returns the service name.

func (*Handler) Purge

func (h *Handler) Purge(ctx context.Context, cutoff time.Time)

Purge implements service.Purgeable by removing all Kinesis streams older than cutoff.

func (*Handler) Reset

func (h *Handler) Reset()

Reset clears all in-memory state from the backend. It is used by the POST /_gopherstack/reset endpoint for CI pipelines and rapid local development.

func (*Handler) Restore

func (h *Handler) Restore(ctx context.Context, data []byte) error

Restore implements persistence.Persistable by delegating to the backend.

func (*Handler) RouteMatcher

func (h *Handler) RouteMatcher() service.Matcher

RouteMatcher returns a function that matches incoming Kinesis requests.

func (*Handler) Shutdown

func (h *Handler) Shutdown(ctx context.Context)

Shutdown implements service.Shutdowner: it flushes any channel-buffered records to S3 before the process exits, so records accepted since the last interval flush are not lost (this backend does not persist buffered records across a snapshot/restore cycle -- see PARITY.md). If ctx expires before the flush finishes, Shutdown returns immediately.

func (*Handler) Snapshot

func (h *Handler) Snapshot(ctx context.Context) []byte

Snapshot implements persistence.Persistable by delegating to the backend.

func (*Handler) StartWorker

func (h *Handler) StartWorker(ctx context.Context) error

StartWorker starts the background janitor (if configured) and the channel delivery interval flusher (see channel_delivery.go's runChannelFlusher).

func (*Handler) StopWorker

func (h *Handler) StopWorker()

StopWorker stops the background worker if one is configured.

func (*Handler) WithJanitor

func (h *Handler) WithJanitor(interval time.Duration, taskTimeout ...time.Duration) *Handler

WithJanitor attaches a background janitor to the handler. If the backend is not an *InMemoryBackend, this is a no-op.

func (*Handler) WithSubscribeToShardTiming

func (h *Handler) WithSubscribeToShardTiming(streamDuration, pollInterval, heartbeatInterval time.Duration) *Handler

WithSubscribeToShardTiming overrides SubscribeToShard's stream duration, poll interval, and heartbeat interval (see handler_consumers.go). A zero argument keeps that setting's current value, mirroring services/polly's WithStreamLimits zero-means-keep-default pattern.

type InMemoryBackend

type InMemoryBackend struct {
	OnStreamPurged func(string)
	// contains filtered or unexported fields
}

InMemoryBackend implements StorageBackend using in-memory maps.

Streams are keyed by a composite "region/name" key (see streamKey) inside a single flat store.Table, with a secondary store.Index grouping them by region — so same-named streams in different regions remain fully isolated, including their shards, records, and consumers, all of which stay inline fields on Stream (see pkgs/store's package doc: shards/records are the per-stream hot path and are not decomposed into their own tables). fisThroughputFaults and resourcePolicies are NOT store.Table candidates: their value types (a pointer with no self-describing identity, and a bare string) carry no key of their own to hand a Table keyFn, so they remain plain nested maps guarded the same way as before.

func NewInMemoryBackend

func NewInMemoryBackend() *InMemoryBackend

NewInMemoryBackend creates a new empty InMemoryBackend with default account/region.

func NewInMemoryBackendWithConfig

func NewInMemoryBackendWithConfig(accountID, region string) *InMemoryBackend

NewInMemoryBackendWithConfig creates a new InMemoryBackend with the given account ID and region.

func (*InMemoryBackend) AddStreamInternal

func (b *InMemoryBackend) AddStreamInternal(stream *Stream)

AddStreamInternal seeds a stream directly into the backend for testing. Caller must provide a non-nil stream with at least Name and ARN set. The stream is placed in the region encoded in its ARN, falling back to the backend's default region when the ARN carries none.

func (*InMemoryBackend) CountOnDemandStreams added in v1.3.1

func (b *InMemoryBackend) CountOnDemandStreams(ctx context.Context) int

CountOnDemandStreams returns the number of ON_DEMAND streams in the region carried on ctx, for DescribeLimits' required OnDemandStreamCount member (kinesis@v1.46.4 api_op_DescribeLimits.go:34-45). DescribeLimits is region-scoped in AWS, matching CountOpenShards' convention.

func (*InMemoryBackend) CountOpenShards

func (b *InMemoryBackend) CountOpenShards(ctx context.Context) int

CountOpenShards returns the total number of open (non-closed) shards across every stream in the region carried on ctx. DescribeLimits is region-scoped in AWS, so this counts within a single region.

func (*InMemoryBackend) CreateChannel

func (b *InMemoryBackend) CreateChannel(ctx context.Context, input *CreateChannelInput) (*CreateChannelOutput, error)

CreateChannel creates a channel that delivers records from a Kinesis data stream to an S3 or S3-Tables destination (api_op_CreateChannel.go). Records are not actually delivered -- see PARITY.md. CreateChannel is documented as asynchronous (CREATING then ACTIVE); this backend applies it synchronously, matching the precedent already set for UpdateStreamWarmThroughput/UpdateStreamMode.

func (*InMemoryBackend) CreateStream

func (b *InMemoryBackend) CreateStream(ctx context.Context, input *CreateStreamInput) error

CreateStream creates a new Kinesis stream.

func (*InMemoryBackend) DecreaseStreamRetentionPeriod

func (b *InMemoryBackend) DecreaseStreamRetentionPeriod(
	ctx context.Context,
	input *DecreaseStreamRetentionPeriodInput,
) error

DecreaseStreamRetentionPeriod decreases the retention period for a stream. Mirroring IncreaseStreamRetentionPeriod, a target equal to the current retention period is an idempotent no-op returning success (HTTP 200), matching real AWS behaviour. A target above the current period or below minRetentionHours (24h) is rejected with InvalidArgumentException.

func (*InMemoryBackend) DeleteChannel

func (b *InMemoryBackend) DeleteChannel(ctx context.Context, input *DeleteChannelInput) error

DeleteChannel deletes the specified channel (api_op_DeleteChannel.go). Deletion is synchronous -- DeleteChannel's own doc comment, unlike CreateChannel/UpdateChannel's, describes no CREATING/UPDATING-style asynchronous transition, so there is no documented DELETING state to model. Any buffered records are flushed to S3 (best-effort) after the channel row is removed, so records already accepted are not silently dropped.

func (*InMemoryBackend) DeleteResourcePolicy

func (b *InMemoryBackend) DeleteResourcePolicy(ctx context.Context, input *DeleteResourcePolicyInput) error

DeleteResourcePolicy removes the resource-based policy for the given stream or consumer ARN.

func (*InMemoryBackend) DeleteStream

func (b *InMemoryBackend) DeleteStream(ctx context.Context, input *DeleteStreamInput) error

DeleteStream removes a stream, then flushes (best-effort) the buffer of any channel sourced from it, so records already accepted by PutRecord are not silently dropped by the removal (see channel_delivery.go).

func (*InMemoryBackend) DeregisterStreamConsumer

func (b *InMemoryBackend) DeregisterStreamConsumer(ctx context.Context, input *DeregisterStreamConsumerInput) error

DeregisterStreamConsumer removes a registered consumer from a stream.

func (*InMemoryBackend) DescribeAccountSettings

func (b *InMemoryBackend) DescribeAccountSettings(_ context.Context) (*DescribeAccountSettingsOutput, error)

DescribeAccountSettings returns the account's minimum throughput billing commitment configuration (kinesis@v1.46.4 api_op_DescribeAccountSettings.go:34-45).

func (*InMemoryBackend) DescribeChannel

func (b *InMemoryBackend) DescribeChannel(
	_ context.Context,
	input *DescribeChannelInput,
) (*DescribeChannelOutput, error)

DescribeChannel describes the specified channel (api_op_DescribeChannel.go).

func (*InMemoryBackend) DescribeStream

func (b *InMemoryBackend) DescribeStream(
	ctx context.Context,
	input *DescribeStreamInput,
) (*DescribeStreamOutput, error)

DescribeStream returns full stream details including shards.

func (*InMemoryBackend) DescribeStreamConsumer

func (b *InMemoryBackend) DescribeStreamConsumer(
	ctx context.Context,
	input *DescribeStreamConsumerInput,
) (*DescribeStreamConsumerOutput, error)

DescribeStreamConsumer returns details about a registered consumer. Lookup is by ConsumerARN, or by StreamARN + ConsumerName.

func (*InMemoryBackend) DisableEnhancedMonitoring

DisableEnhancedMonitoring removes shard-level metrics from a stream.

func (*InMemoryBackend) EnableEnhancedMonitoring

func (b *InMemoryBackend) EnableEnhancedMonitoring(
	ctx context.Context,
	input *EnableEnhancedMonitoringInput,
) (*EnableEnhancedMonitoringOutput, error)

EnableEnhancedMonitoring adds shard-level metrics to a stream.

func (*InMemoryBackend) FlushAllChannels

func (b *InMemoryBackend) FlushAllChannels(ctx context.Context)

FlushAllChannels forces immediate delivery of every channel's buffered records. Used by Handler.Shutdown and tests.

func (*InMemoryBackend) FlushChannel

func (b *InMemoryBackend) FlushChannel(ctx context.Context, channelARN string)

FlushChannel forces immediate delivery of channelARN's buffered records, regardless of its DataFreshnessInSeconds window. Used by DeleteChannel, graceful shutdown, and tests (this backend has no injectable clock for interval-based flush -- see PARITY.md).

func (*InMemoryBackend) GetRecords

func (b *InMemoryBackend) GetRecords(ctx context.Context, input *GetRecordsInput) (*GetRecordsOutput, error)

GetRecords retrieves records starting at the given shard iterator position.

The region is taken from the iterator token (encoded by GetShardIterator), not from ctx, so an iterator issued for one region always reads that region's records even if the GetRecords call carries a different ctx region.

func (*InMemoryBackend) GetResourcePolicy

func (b *InMemoryBackend) GetResourcePolicy(
	ctx context.Context,
	input *GetResourcePolicyInput,
) (*GetResourcePolicyOutput, error)

GetResourcePolicy retrieves the resource-based policy for the given stream or consumer ARN.

func (*InMemoryBackend) GetShardIterator

func (b *InMemoryBackend) GetShardIterator(
	ctx context.Context,
	input *GetShardIteratorInput,
) (*GetShardIteratorOutput, error)

GetShardIterator returns an iterator for reading records from a shard.

func (*InMemoryBackend) IncreaseStreamRetentionPeriod

func (b *InMemoryBackend) IncreaseStreamRetentionPeriod(
	ctx context.Context,
	input *IncreaseStreamRetentionPeriodInput,
) error

IncreaseStreamRetentionPeriod increases the retention period for a stream. A target equal to the current retention period is treated as an idempotent no-op returning success (HTTP 200), matching real AWS behaviour: the Terraform AWS provider calls IncreaseStreamRetentionPeriod on stream create for ANY configured retention_period > 0 (see the provider's resourceStreamCreate, guard `v.(int) > 0`), so a stream created with the default retention_period of 24h receives IncreaseStreamRetentionPeriod(24) against a stream already at 24h. Rejecting that equal value with InvalidArgumentException (as a strict reading of the SDK doc "Must be more than the current retention period" would suggest) breaks every default-retention Terraform stream, so real AWS accepts it. A target below the current period, below minRetentionHours (24h), or above maxRetentionHours (8760h) is rejected with InvalidArgumentException.

func (*InMemoryBackend) ListAll

func (b *InMemoryBackend) ListAll(_ context.Context) []StreamInfo

ListAll returns a snapshot of all streams as StreamInfo values across every region. It is used by the dashboard, which presents a global inventory.

func (*InMemoryBackend) ListChannels

func (b *InMemoryBackend) ListChannels(ctx context.Context, input *ListChannelsInput) (*ListChannelsOutput, error)

ListChannels lists the channels in the caller's account/region, optionally filtered by source stream (api_op_ListChannels.go).

func (*InMemoryBackend) ListShards

func (b *InMemoryBackend) ListShards(ctx context.Context, input *ListShardsInput) (*ListShardsOutput, error)

ListShards returns the shards for a stream.

func (*InMemoryBackend) ListStreamConsumers

func (b *InMemoryBackend) ListStreamConsumers(
	ctx context.Context,
	input *ListStreamConsumersInput,
) (*ListStreamConsumersOutput, error)

ListStreamConsumers lists all registered consumers for a stream.

func (*InMemoryBackend) ListStreams

func (b *InMemoryBackend) ListStreams(ctx context.Context, input *ListStreamsInput) (*ListStreamsOutput, error)

ListStreams returns stream names with optional pagination.

AWS contract: results are returned in alphabetical order. When `Limit` is set the response contains at most that many names. Pagination is keyed on either `ExclusiveStartStreamName` or the opaque `NextToken` (which we treat as the previously returned last stream name) so that callers can iterate over arbitrarily large account inventories.

func (*InMemoryBackend) ListTagsForResource

func (b *InMemoryBackend) ListTagsForResource(
	ctx context.Context,
	input *ListTagsForResourceInput,
) (*ListTagsForResourceOutput, error)

ListTagsForResource returns the tags associated with a Kinesis resource identified by its ARN -- a stream (tags stored on the stream's internal Tags store, set via TagResource) or, per the real op's doc comment ("the specified Kinesis resource"), an enhanced fan-out consumer (tags stored on Consumer.Tags, set via RegisterStreamConsumer's Tags parameter or TagResource against the consumer's own ARN).

func (*InMemoryBackend) MergeShards

func (b *InMemoryBackend) MergeShards(ctx context.Context, input *MergeShardsInput) error

MergeShards merges two adjacent shards into one. The merged shard spans the combined hash key range of both parent shards.

func (*InMemoryBackend) OnDemandStreamCountLimit added in v1.3.1

func (b *InMemoryBackend) OnDemandStreamCountLimit(_ context.Context) int

OnDemandStreamCountLimit returns the account's current cap on ON_DEMAND streams, for DescribeLimits' required OnDemandStreamCountLimit member. Real AWS manages this as a Service Quota, not adjustable via UpdateAccountSettings; see SetOnDemandStreamCountLimit for how this backend exposes changing it (a Go-level config knob, not a wire operation).

func (*InMemoryBackend) Purge

func (b *InMemoryBackend) Purge(ctx context.Context, cutoff time.Time)

Purge removes all Kinesis streams and consumers created before the cutoff time.

func (*InMemoryBackend) PutRecord

func (b *InMemoryBackend) PutRecord(ctx context.Context, input *PutRecordInput) (*PutRecordOutput, error)

PutRecord writes a single record to a stream shard, then delivers it to any ACTIVE channel sourced from the stream (see deliverPutToChannels).

func (*InMemoryBackend) PutRecords

func (b *InMemoryBackend) PutRecords(ctx context.Context, input *PutRecordsInput) (*PutRecordsOutput, error)

PutRecords writes multiple records to a stream.

Request-level validation errors (empty/oversized batch, unknown stream) fail the whole call with a single top-level exception, matching the AWS contract: only per-record issues (throughput, per-record validation) surface as per-entry ErrorCode/ErrorMessage inside a 200 response.

func (*InMemoryBackend) PutResourcePolicy

func (b *InMemoryBackend) PutResourcePolicy(ctx context.Context, input *PutResourcePolicyInput) error

PutResourcePolicy stores a resource-based policy for the given stream or consumer ARN.

func (*InMemoryBackend) Region

func (b *InMemoryBackend) Region() string

Region returns the AWS region this backend is configured to use as its default.

func (*InMemoryBackend) RegisterStreamConsumer

func (b *InMemoryBackend) RegisterStreamConsumer(
	ctx context.Context,
	input *RegisterStreamConsumerInput,
) (*RegisterStreamConsumerOutput, error)

RegisterStreamConsumer registers a new enhanced fan-out consumer on a stream.

func (*InMemoryBackend) Reset

func (b *InMemoryBackend) Reset()

Reset clears all in-memory state from the backend. It is used by the POST /_gopherstack/reset endpoint for CI pipelines and rapid local development.

func (*InMemoryBackend) Restore

func (b *InMemoryBackend) Restore(ctx context.Context, data []byte) error

Restore loads backend state from a JSON snapshot. It implements persistence.Persistable.

func (*InMemoryBackend) SetOnDemandStreamCountLimit added in v1.3.1

func (b *InMemoryBackend) SetOnDemandStreamCountLimit(n int)

SetOnDemandStreamCountLimit configures the account-level cap CreateStream enforces for ON_DEMAND streams (default defaultOnDemandStreamCountLimit). No real Kinesis wire operation can change this account setting -- it was previously (and incorrectly) exposed as a fabricated field on UpdateAccountSettingsInput; real AWS manages it as a Service Quota. This method is the Go-level replacement, mirroring how WithKMSValidator wires cross-service config outside the wire protocol.

func (*InMemoryBackend) SetS3Writer

func (b *InMemoryBackend) SetS3Writer(w ChannelS3Writer)

SetS3Writer wires the S3 backend used to deliver channel-buffered records to their configured S3DestinationConfiguration bucket. See cli.go's wireKinesisS3Delivery.

func (*InMemoryBackend) Snapshot

func (b *InMemoryBackend) Snapshot(ctx context.Context) []byte

Snapshot serialises the backend state to JSON. It implements persistence.Persistable. Note: shard sequence number counters are now serialised via the NextSeq field.

func (*InMemoryBackend) SplitShard

func (b *InMemoryBackend) SplitShard(ctx context.Context, input *SplitShardInput) error

SplitShard splits a shard into two at the given new starting hash key.

func (*InMemoryBackend) StartStreamEncryption

func (b *InMemoryBackend) StartStreamEncryption(ctx context.Context, input *StartStreamEncryptionInput) error

StartStreamEncryption enables server-side encryption on a stream.

func (*InMemoryBackend) StopStreamEncryption

func (b *InMemoryBackend) StopStreamEncryption(ctx context.Context, input *StopStreamEncryptionInput) error

StopStreamEncryption disables server-side encryption on a stream.

func (*InMemoryBackend) SubscribeToShard

func (b *InMemoryBackend) SubscribeToShard(
	ctx context.Context,
	input *SubscribeToShardInput,
) (*SubscribeToShardOutput, error)

SubscribeToShard delivers records from a shard to an enhanced fan-out consumer. For mock purposes this is a single-shot delivery of all available records.

func (*InMemoryBackend) TagResource

func (b *InMemoryBackend) TagResource(ctx context.Context, input *TagResourceInput) error

TagResource adds or updates tags on a Kinesis resource identified by its ARN -- a stream (the ARN-based counterpart to AddTagsToStream) or an enhanced fan-out consumer.

func (*InMemoryBackend) TaggedStreams added in v1.2.0

func (b *InMemoryBackend) TaggedStreams() []TaggedEntry

TaggedStreams returns every Kinesis stream that currently has at least one tag, across every region this backend holds streams for.

func (*InMemoryBackend) UntagResource

func (b *InMemoryBackend) UntagResource(ctx context.Context, input *UntagResourceInput) error

UntagResource removes tags from a Kinesis resource identified by its ARN -- a stream (the ARN-based counterpart to RemoveTagsFromStream) or an enhanced fan-out consumer.

func (*InMemoryBackend) UpdateAccountSettings

UpdateAccountSettings sets the account's minimum throughput billing commitment status (kinesis@v1.46.4 api_op_UpdateAccountSettings.go:42-51). This backend has no billing engine: no billing behaviour follows from enabling the commitment. Status/StartedAt/EndedAt only track the requested transition; see MinimumThroughputBillingCommitmentOutput.

func (*InMemoryBackend) UpdateChannel

UpdateChannel updates a channel's data-freshness interval or CloudWatch Logs configuration (api_op_UpdateChannel.go). UpdateChannel is documented as asynchronous (UPDATING then ACTIVE); this backend applies it synchronously, matching CreateChannel's disclosed simplification.

func (*InMemoryBackend) UpdateMaxRecordSize

func (b *InMemoryBackend) UpdateMaxRecordSize(ctx context.Context, input *UpdateMaxRecordSizeInput) error

UpdateMaxRecordSize changes the per-record data payload size limit for a stream. The wire input is MaxRecordSizeInKiB (kinesis@v1.46.4 api_op_UpdateMaxRecordSize.go:30-47); this backend stores the limit in bytes on Stream.MaxRecordSizeBytes, so the requested KiB value is converted via bytesPerKiB. The valid range is [defaultMaxRecordSizeBytes, absoluteMaxRecordSizeBytes] (1 MiB - 10 MiB). The real Input has no StreamName member, only StreamARN.

func (*InMemoryBackend) UpdateShardCount

func (b *InMemoryBackend) UpdateShardCount(
	ctx context.Context,
	input *UpdateShardCountInput,
) (*UpdateShardCountOutput, error)

UpdateShardCount resizes a stream to the given number of shards. Existing records in the stream are not migrated; new shards start empty.

func (*InMemoryBackend) UpdateStreamMode

func (b *InMemoryBackend) UpdateStreamMode(ctx context.Context, input *UpdateStreamModeInput) error

UpdateStreamMode changes the mode of a stream identified by its ARN.

func (*InMemoryBackend) UpdateStreamWarmThroughput

UpdateStreamWarmThroughput configures pre-warmed throughput for a stream (kinesis@v1.46.4 api_op_UpdateStreamWarmThroughput.go:63-70, required WarmThroughputMiBps). Real AWS applies this asynchronously (stream goes UPDATING then back to ACTIVE); this backend has no transient-state model for that (streams are always ACTIVE), so the change is applied synchronously and Current/Target always match on read -- see UpdateStreamWarmThroughputOutput and PARITY.md.

func (*InMemoryBackend) WithClock

func (b *InMemoryBackend) WithClock(now func() time.Time) *InMemoryBackend

WithClock overrides the backend's time source, used by tests to drive AT_TRIM_HORIZON/AT_TIMESTAMP retention math and ON_DEMAND write-throughput auto-scaling deterministically -- no time.Sleep, no real wall-clock waits. Mirrors services/polly/throttle.go's WithClock.

func (*InMemoryBackend) WithKMSValidator added in v1.2.0

func (b *InMemoryBackend) WithKMSValidator(v KMSKeyValidator) *InMemoryBackend

WithKMSValidator attaches a KMSKeyValidator so StartStreamEncryption can verify a KeyId resolves to a real, usable KMS key. Returns b for chaining.

type IncreaseStreamRetentionPeriodInput

type IncreaseStreamRetentionPeriodInput struct {
	StreamName           string
	RetentionPeriodHours int
}

IncreaseStreamRetentionPeriodInput is the input for IncreaseStreamRetentionPeriod.

type Janitor

type Janitor struct {
	Backend *InMemoryBackend

	Interval time.Duration
	// TaskTimeout bounds each individual janitor task. When non-zero, each task
	// runs with a child context that expires after this duration, preventing a
	// stalled operation from blocking the janitor loop indefinitely.
	TaskTimeout time.Duration
	// contains filtered or unexported fields
}

Janitor is the Kinesis background worker that enforces per-stream retention periods by evicting records older than stream.RetentionPeriod hours.

func NewJanitor

func NewJanitor(backend *InMemoryBackend, interval time.Duration) *Janitor

NewJanitor creates a new Kinesis Janitor for the given backend. A zero interval falls back to defaultJanitorInterval.

func (*Janitor) Run

func (j *Janitor) Run(ctx context.Context)

Run runs the janitor loop until ctx is cancelled.

func (*Janitor) Stop

func (j *Janitor) Stop()

Stop explicitly shuts down the janitor.

func (*Janitor) SweepOnce

func (j *Janitor) SweepOnce(ctx context.Context)

SweepOnce executes a single retention sweep. Exposed for testing.

type KMSKeyValidator added in v1.2.0

type KMSKeyValidator interface {
	// ValidateKMSKey resolves keyID against the KMS backend and returns nil if
	// the key exists and is usable, or one of ErrKMSNotFound/ErrKMSDisabled/
	// ErrKMSInvalidState (kinesis sentinels) describing why it is not.
	ValidateKMSKey(ctx context.Context, keyID string) error
}

KMSKeyValidator optionally validates a KMS KeyId against a real KMS backend. Implemented by an adapter wrapping the KMS service backend and attached via WithKMSValidator (mirrors services/ssm's KMSEncryptor injection pattern -- see cli.go's wireKinesisKMS). When no validator is wired, StartStreamEncryption still enforces the KeyId *shape* AWS documents (UUID / key ARN / alias ARN / "alias/..." name) but cannot know whether the key actually exists, is disabled, or is pending deletion -- those KMS-specific exceptions require real cross-service key state.

type ListChannelsInput

type ListChannelsInput struct {
	NextToken    string
	StreamFilter []ChannelStreamFilter
	MaxResults   int
}

ListChannelsInput is the input for ListChannels.

type ListChannelsOutput

type ListChannelsOutput struct {
	NextToken        string
	ChannelSummaries []Channel
}

ListChannelsOutput is the output for ListChannels.

type ListShardsInput

type ListShardsInput struct {
	ShardFilterTimestamp  *time.Time
	StreamName            string
	NextToken             string
	ExclusiveStartShardID string
	ShardFilter           string
	ShardFilterType       string
	ShardFilterShardID    string
	MaxResults            int
}

ListShardsInput is the input for ListShards.

type ListShardsOutput

type ListShardsOutput struct {
	NextToken string
	Shards    []ShardDescription
}

ListShardsOutput is the output for ListShards.

type ListStreamConsumersInput

type ListStreamConsumersInput struct {
	StreamARN  string
	NextToken  string
	MaxResults int
}

ListStreamConsumersInput is the input for ListStreamConsumers.

type ListStreamConsumersOutput

type ListStreamConsumersOutput struct {
	NextToken string
	Consumers []Consumer
}

ListStreamConsumersOutput is the output for ListStreamConsumers.

type ListStreamsInput

type ListStreamsInput struct {
	NextToken                string
	ExclusiveStartStreamName string
	Limit                    int
}

ListStreamsInput is the input for ListStreams.

type ListStreamsOutput

type ListStreamsOutput struct {
	NextToken      string
	StreamNames    []string
	HasMoreStreams bool
}

ListStreamsOutput is the output for ListStreams.

type ListTagsForResourceInput

type ListTagsForResourceInput struct {
	ResourceARN string
}

ListTagsForResourceInput is the input for ListTagsForResource.

type ListTagsForResourceOutput

type ListTagsForResourceOutput struct {
	Tags map[string]string
}

ListTagsForResourceOutput is the output for ListTagsForResource.

type MergeShardsInput

type MergeShardsInput struct {
	StreamName           string
	StreamARN            string
	ShardToMerge         string
	AdjacentShardToMerge string
}

MergeShardsInput is the input for MergeShards.

type MinimumThroughputBillingCommitmentInput added in v1.3.1

type MinimumThroughputBillingCommitmentInput struct {
	// Status is required: minimumThroughputBillingCommitmentEnabled or
	// minimumThroughputBillingCommitmentDisabled.
	Status string
}

MinimumThroughputBillingCommitmentInput is the input shape for the commitment status requested via UpdateAccountSettings (types.MinimumThroughputBillingCommitmentInput; Status is its only, required, member -- kinesis@v1.46.4 types/types.go:168-176).

type MinimumThroughputBillingCommitmentOutput added in v1.3.1

type MinimumThroughputBillingCommitmentOutput struct {
	EarliestAllowedEndAt time.Time `json:"earliestAllowedEndAt"`
	EndedAt              time.Time `json:"endedAt"`
	StartedAt            time.Time `json:"startedAt"`
	Status               string    `json:"status"`
}

MinimumThroughputBillingCommitmentOutput is the account's current minimum throughput billing commitment (types.MinimumThroughputBillingCommitmentOutput, kinesis@v1.46.4 types/types.go:178-197). This backend has no billing engine: Status/StartedAt/EndedAt only track the state transitions UpdateAccountSettings requests; EarliestAllowedEndAt is never populated since computing it needs a commitment-window model this backend doesn't have (see PARITY.md gaps), and Status never reports minimumThroughputBillingCommitmentEnabledUntilEnd for the same reason.

type Provider

type Provider struct{}

Provider implements service.Provider for the Kinesis service.

func (*Provider) Init

Init initializes the Kinesis service backend and handler.

func (*Provider) Name

func (p *Provider) Name() string

Name returns the service provider name.

type PutRecordInput

type PutRecordInput struct {
	StreamName      string
	PartitionKey    string
	ExplicitHashKey string
	Data            []byte
}

PutRecordInput is the input for PutRecord.

type PutRecordOutput

type PutRecordOutput struct {
	ShardID        string
	SequenceNumber string
	EncryptionType string
}

PutRecordOutput is the output for PutRecord.

type PutRecordsEntry

type PutRecordsEntry struct {
	PartitionKey    string
	ExplicitHashKey string
	Data            []byte
}

PutRecordsEntry is a single entry in a PutRecords request.

type PutRecordsInput

type PutRecordsInput struct {
	StreamName string
	Records    []PutRecordsEntry
}

PutRecordsInput is the input for PutRecords.

type PutRecordsOutput

type PutRecordsOutput struct {
	Records           []PutRecordsResultEntry
	FailedRecordCount int
}

PutRecordsOutput is the output for PutRecords.

type PutRecordsResultEntry

type PutRecordsResultEntry struct {
	ShardID        string
	SequenceNumber string
	ErrorCode      string
	ErrorMessage   string
}

PutRecordsResultEntry is a single result entry in a PutRecords response.

type PutResourcePolicyInput

type PutResourcePolicyInput struct {
	ResourceARN string
	Policy      string
}

PutResourcePolicyInput is the input for PutResourcePolicy.

type Record

type Record struct {
	ApproximateArrivalTimestamp time.Time `json:"approximateArrivalTimestamp"`
	PartitionKey                string    `json:"partitionKey"`
	SequenceNumber              string    `json:"sequenceNumber"`
	Data                        []byte    `json:"data"`
}

Record represents a single Kinesis data record.

type RegisterStreamConsumerInput

type RegisterStreamConsumerInput struct {
	Tags         map[string]string
	StreamARN    string
	ConsumerName string
}

RegisterStreamConsumerInput is the input for RegisterStreamConsumer.

type RegisterStreamConsumerOutput

type RegisterStreamConsumerOutput struct {
	Consumer Consumer
}

RegisterStreamConsumerOutput is the output for RegisterStreamConsumer.

type Settings

type Settings struct {
	JanitorInterval time.Duration `json:"janitor_interval" env:"KINESIS_JANITOR_INTERVAL" default:"1m" help:"Janitor tick interval."` //nolint:lll // Kong struct tag makes this line long
}

Settings holds service-level configuration for the Kinesis backend. Fields are picked up by the Kong CLI parser when this struct is embedded in the root CLI command.

type Shard

type Shard struct {
	// StartedAt is when this shard became open (stream creation for the initial
	// shard set, or reshard time for shards born from SplitShard/MergeShards/
	// UpdateShardCount/UpdateStreamMode). Used by ListShards' AT_TIMESTAMP/
	// FROM_TIMESTAMP/AT_TRIM_HORIZON ShardFilter to bound shard lineage by time.
	StartedAt time.Time `json:"startedAt"`
	// ClosedAt is when this shard was closed (zero if still open). Populated
	// alongside Closed by closeShard. omitempty has no effect on a struct
	// field like time.Time, so it is intentionally omitted here.
	ClosedAt              time.Time    `json:"closedAt"`
	ID                    string       `json:"id"`
	HashKeyRangeStart     string       `json:"hashKeyRangeStart"`
	HashKeyRangeEnd       string       `json:"hashKeyRangeEnd"`
	ParentShardID         string       `json:"parentShardId,omitempty"`
	AdjacentParentShardID string       `json:"adjacentParentShardId,omitempty"`
	Records               shardRecords `json:"records"`
	NextSeq               uint64       `json:"nextSeq"`
	Closed                bool         `json:"closed,omitempty"`
}

Shard represents a single Kinesis shard within a stream.

type ShardDescription

type ShardDescription struct {
	ShardID                  string
	HashKeyRangeStart        string
	HashKeyRangeEnd          string
	SequenceNumberRangeStart string
	SequenceNumberRangeEnd   string
	ParentShardID            string
	AdjacentParentShardID    string
	Closed                   bool
}

ShardDescription describes a shard in a DescribeStream response.

type ShardIterator

type ShardIterator struct {
	CreatedAt      time.Time `json:"CreatedAt"`
	StreamName     string    `json:"StreamName"`
	ShardID        string    `json:"ShardID"`
	SequenceNumber string    `json:"SequenceNumber"`
	Region         string    `json:"Region"`
	Position       int       `json:"Position"`
}

ShardIterator holds the position within a shard for GetRecords. Region is encoded into the iterator token so that GetRecords resolves the record store of the same region the iterator was issued in, keeping same-named streams in different regions isolated on the record hot path.

type SplitShardInput

type SplitShardInput struct {
	StreamName         string
	StreamARN          string
	ShardToSplit       string
	NewStartingHashKey string
}

SplitShardInput is the input for SplitShard.

type StartStreamEncryptionInput

type StartStreamEncryptionInput struct {
	StreamName     string
	StreamARN      string
	EncryptionType string
	KeyID          string
}

StartStreamEncryptionInput is the input for StartStreamEncryption.

type StartingPosition

type StartingPosition struct {
	Timestamp      *time.Time `json:"Timestamp,omitempty"`
	Type           string     `json:"Type"`
	SequenceNumber string     `json:"SequenceNumber,omitempty"`
}

StartingPosition describes where to start reading in SubscribeToShard.

type StopStreamEncryptionInput

type StopStreamEncryptionInput struct {
	StreamName     string
	StreamARN      string
	EncryptionType string
	KeyID          string
}

StopStreamEncryptionInput is the input for StopStreamEncryption.

type StorageBackend

type StorageBackend interface {
	CreateStream(ctx context.Context, input *CreateStreamInput) error
	DeleteStream(ctx context.Context, input *DeleteStreamInput) error
	DescribeStream(ctx context.Context, input *DescribeStreamInput) (*DescribeStreamOutput, error)
	ListStreams(ctx context.Context, input *ListStreamsInput) (*ListStreamsOutput, error)
	PutRecord(ctx context.Context, input *PutRecordInput) (*PutRecordOutput, error)
	PutRecords(ctx context.Context, input *PutRecordsInput) (*PutRecordsOutput, error)
	GetShardIterator(ctx context.Context, input *GetShardIteratorInput) (*GetShardIteratorOutput, error)
	GetRecords(ctx context.Context, input *GetRecordsInput) (*GetRecordsOutput, error)
	ListShards(ctx context.Context, input *ListShardsInput) (*ListShardsOutput, error)
	RegisterStreamConsumer(
		ctx context.Context,
		input *RegisterStreamConsumerInput,
	) (*RegisterStreamConsumerOutput, error)
	DescribeStreamConsumer(
		ctx context.Context,
		input *DescribeStreamConsumerInput,
	) (*DescribeStreamConsumerOutput, error)
	ListStreamConsumers(ctx context.Context, input *ListStreamConsumersInput) (*ListStreamConsumersOutput, error)
	DeregisterStreamConsumer(ctx context.Context, input *DeregisterStreamConsumerInput) error
	SubscribeToShard(ctx context.Context, input *SubscribeToShardInput) (*SubscribeToShardOutput, error)
	UpdateShardCount(ctx context.Context, input *UpdateShardCountInput) (*UpdateShardCountOutput, error)
	EnableEnhancedMonitoring(
		ctx context.Context,
		input *EnableEnhancedMonitoringInput,
	) (*EnableEnhancedMonitoringOutput, error)
	DisableEnhancedMonitoring(
		ctx context.Context,
		input *DisableEnhancedMonitoringInput,
	) (*DisableEnhancedMonitoringOutput, error)
	IncreaseStreamRetentionPeriod(ctx context.Context, input *IncreaseStreamRetentionPeriodInput) error
	DecreaseStreamRetentionPeriod(ctx context.Context, input *DecreaseStreamRetentionPeriodInput) error
	MergeShards(ctx context.Context, input *MergeShardsInput) error
	SplitShard(ctx context.Context, input *SplitShardInput) error
	StartStreamEncryption(ctx context.Context, input *StartStreamEncryptionInput) error
	StopStreamEncryption(ctx context.Context, input *StopStreamEncryptionInput) error
	DeleteResourcePolicy(ctx context.Context, input *DeleteResourcePolicyInput) error
	GetResourcePolicy(ctx context.Context, input *GetResourcePolicyInput) (*GetResourcePolicyOutput, error)
	PutResourcePolicy(ctx context.Context, input *PutResourcePolicyInput) error
	ListTagsForResource(ctx context.Context, input *ListTagsForResourceInput) (*ListTagsForResourceOutput, error)
	TagResource(ctx context.Context, input *TagResourceInput) error
	UntagResource(ctx context.Context, input *UntagResourceInput) error
	UpdateStreamMode(ctx context.Context, input *UpdateStreamModeInput) error
	UpdateAccountSettings(ctx context.Context, input *UpdateAccountSettingsInput) (*UpdateAccountSettingsOutput, error)
	UpdateMaxRecordSize(ctx context.Context, input *UpdateMaxRecordSizeInput) error
	UpdateStreamWarmThroughput(
		ctx context.Context,
		input *UpdateStreamWarmThroughputInput,
	) (*UpdateStreamWarmThroughputOutput, error)
	DescribeAccountSettings(ctx context.Context) (*DescribeAccountSettingsOutput, error)
	CreateChannel(ctx context.Context, input *CreateChannelInput) (*CreateChannelOutput, error)
	DeleteChannel(ctx context.Context, input *DeleteChannelInput) error
	DescribeChannel(ctx context.Context, input *DescribeChannelInput) (*DescribeChannelOutput, error)
	ListChannels(ctx context.Context, input *ListChannelsInput) (*ListChannelsOutput, error)
	UpdateChannel(ctx context.Context, input *UpdateChannelInput) (*UpdateChannelOutput, error)
	CountOpenShards(ctx context.Context) int
	CountOnDemandStreams(ctx context.Context) int
	OnDemandStreamCountLimit(ctx context.Context) int
	ListAll(ctx context.Context) []StreamInfo
}

StorageBackend defines the interface for a Kinesis backend.

Every method takes a context.Context so the per-request AWS region can be threaded through and resources kept isolated per region. The region is read from the context via getRegion, falling back to the backend's default region when the context carries no region.

type Stream

type Stream struct {
	CreatedAt time.Time `json:"createdAt"`

	Tags      *tags.Tags           `json:"tags,omitempty"`
	Consumers map[string]*Consumer `json:"consumers,omitempty"`
	Name      string               `json:"name"`
	ARN       string               `json:"arn"`
	// Region is the AWS region this stream lives in. It is the second half of
	// the composite key (see streamKey in backend.go) that keeps same-named
	// streams in different regions isolated inside the single flat
	// store.Table[Stream] — the region-nested map it replaced used the
	// region as an outer map key instead of a field on Stream itself.
	Region             string   `json:"region,omitempty"`
	Status             string   `json:"status"`
	EncryptionType     string   `json:"encryptionType,omitempty"`
	KeyID              string   `json:"keyId,omitempty"`
	StreamMode         string   `json:"streamMode,omitempty"`
	Shards             []*Shard `json:"shards"`
	EnhancedMonitoring []string `json:"enhancedMonitoring,omitempty"`
	RetentionPeriod    int      `json:"retentionPeriod"`
	// MaxRecordSizeBytes is the per-record data payload size limit for this stream.
	// Defaults to defaultMaxRecordSizeBytes (1 MiB); updatable via UpdateMaxRecordSize
	// (wire unit is MaxRecordSizeInKiB; converted to bytes on write via bytesPerKiB).
	MaxRecordSizeBytes int `json:"maxRecordSizeBytes,omitempty"`
	// WarmThroughputMiBps is the stream's current UpdateStreamWarmThroughput
	// setting. Applied synchronously (this backend has no UPDATING transient
	// state), so Current and Target always match on read -- see
	// UpdateStreamWarmThroughputOutput and PARITY.md.
	WarmThroughputMiBps int `json:"warmThroughputMiBps,omitempty"`
	// contains filtered or unexported fields
}

Stream represents an in-memory Kinesis stream.

type StreamInfo

type StreamInfo struct {
	Name       string
	ARN        string
	Status     string
	ShardCount int
}

StreamInfo holds summary information about a stream, safe to return without lock.

type StreamModeDetails

type StreamModeDetails struct {
	StreamMode string
}

StreamModeDetails describes the mode of a Kinesis stream.

type SubscribeToShardEvent

type SubscribeToShardEvent struct {
	ContinuationSequenceNumber string
	Records                    []GetRecordResult
	MillisBehindLatest         int64
}

SubscribeToShardEvent is a single event in the SubscribeToShard response.

type SubscribeToShardInput

type SubscribeToShardInput struct {
	ConsumerARN      string
	ShardID          string
	StartingPosition StartingPosition
}

SubscribeToShardInput is the input for SubscribeToShard.

type SubscribeToShardOutput

type SubscribeToShardOutput struct {
	Event SubscribeToShardEvent
}

SubscribeToShardOutput is the output for SubscribeToShard.

type TagResourceInput

type TagResourceInput struct {
	Tags        map[string]string
	ResourceARN string
}

TagResourceInput is the input for TagResource (ARN-based tagging).

type TaggedEntry added in v1.2.0

type TaggedEntry struct {
	Tags map[string]string
	ARN  string
}

TaggedEntry pairs a stream ARN with its tag map, for cross-service tag enumeration by the Resource Groups Tagging API (see cli.go's wireTaggingKinesis).

type UntagResourceInput

type UntagResourceInput struct {
	ResourceARN string
	TagKeys     []string
}

UntagResourceInput is the input for UntagResource (ARN-based tag removal).

type UpdateAccountSettingsInput

type UpdateAccountSettingsInput struct {
	// MinimumThroughputBillingCommitment is required.
	MinimumThroughputBillingCommitment *MinimumThroughputBillingCommitmentInput
}

UpdateAccountSettingsInput is the input for UpdateAccountSettings.

type UpdateAccountSettingsOutput added in v1.3.1

type UpdateAccountSettingsOutput struct {
	MinimumThroughputBillingCommitment MinimumThroughputBillingCommitmentOutput
}

UpdateAccountSettingsOutput is the output for UpdateAccountSettings.

type UpdateChannelInput

type UpdateChannelInput struct {
	LoggingConfiguration             *ChannelCloudWatchLogsConfig
	S3DestinationConfiguration       *ChannelS3Destination
	S3TablesDestinationConfiguration *ChannelS3TablesDestination
	ChannelARN                       string
}

UpdateChannelInput is the input for UpdateChannel. Per the real op's doc comment, only LoggingConfiguration and the active destination's DataFreshnessInSeconds can be changed: "You cannot change the destination, source stream, record format, schema, encryption configuration, or service execution role of an existing channel".

type UpdateChannelOutput

type UpdateChannelOutput struct {
	ChannelDescription Channel
}

UpdateChannelOutput is the output for UpdateChannel.

type UpdateMaxRecordSizeInput

type UpdateMaxRecordSizeInput struct {
	StreamARN string
	// MaxRecordSizeInKiB is required; wire unit is KiB, not bytes.
	MaxRecordSizeInKiB int
}

UpdateMaxRecordSizeInput is the input for UpdateMaxRecordSize. Unlike most stream-identifying inputs in this file, the real shape has no StreamName member -- only StreamARN (and StreamId, reserved for future use) -- kinesis@v1.46.4 api_op_UpdateMaxRecordSize.go:30-47.

type UpdateShardCountInput

type UpdateShardCountInput struct {
	StreamName       string
	ScalingType      string
	TargetShardCount int
}

UpdateShardCountInput is the input for UpdateShardCount.

type UpdateShardCountOutput

type UpdateShardCountOutput struct {
	StreamName        string
	CurrentShardCount int
	TargetShardCount  int
}

UpdateShardCountOutput is the output for UpdateShardCount.

type UpdateStreamModeInput

type UpdateStreamModeInput struct {
	StreamARN         string
	StreamModeDetails StreamModeDetails
	// WarmThroughputMiBps mirrors UpdateStreamModeInput's own field
	// (kinesis@v1.46.4 api_op_UpdateStreamMode.go, "only valid when the
	// stream mode is being updated to on-demand"). Zero means "not
	// specified".
	WarmThroughputMiBps int
}

UpdateStreamModeInput is the input for UpdateStreamMode.

type UpdateStreamWarmThroughputInput

type UpdateStreamWarmThroughputInput struct {
	StreamName string
	StreamARN  string
	// WarmThroughputMiBps is required (api_op_UpdateStreamWarmThroughput.go:63-70).
	WarmThroughputMiBps int
}

UpdateStreamWarmThroughputInput is the input for UpdateStreamWarmThroughput.

type UpdateStreamWarmThroughputOutput added in v1.3.1

type UpdateStreamWarmThroughputOutput struct {
	StreamARN      string
	StreamName     string
	WarmThroughput WarmThroughputObject
}

UpdateStreamWarmThroughputOutput is the output for UpdateStreamWarmThroughput.

type WarmThroughputObject added in v1.3.1

type WarmThroughputObject struct {
	CurrentMiBps int
	TargetMiBps  int
}

WarmThroughputObject mirrors types.WarmThroughputObject (kinesis@v1.46.4 types/types.go:729-740).

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL