AWS SQS Consume

AWS SQS Consume
Certified
Consume messages from SQS

Polls a queue until maxRecords or maxDuration is reached, stores messages to internal storage, then optionally deletes them from the queue.

yaml
type: io.kestra.plugin.aws.sqs.Consume

Examples

yaml
id: aws_sqs_consume
namespace: company.team

tasks:
  - id: consume
    type: io.kestra.plugin.aws.sqs.Consume
    accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('AWS_SECRET_KEY_ID') }}"
    region: "eu-central-1"
    queueUrl: "https://sqs.eu-central-1.amazonaws.com/000000000000/test-queue"
    maxRecords: 10

Properties

queueUrl *string

The SQS queue URL. The queue must already exist

accessKeyId string

Access Key Id in order to connect to AWS

If no credentials are defined, we will use the default credentials provider chain to fetch credentials.

assets

Assets this task consumes as inputs or produces as outputs, for lineage tracking and the asset graph (Enterprise Edition). A flow declaring this property on a task is rejected in the open-source edition.

Definitions

io.kestra.core.models.assets.AssetsDeclaration

assetFailureBehaviorstring

Possible Values

IGNOREFAILWARN

Asset failure behavior

Behavior applied to the task state when a declared asset fails to render, emit, or be persisted (e.g. a lock conflict): FAIL escalates it to FAILED, WARN (default) warns it if it would otherwise succeed, IGNORE leaves the state untouched.

enableAutobooleanstring

Whether to auto-register assets referenced dynamically at runtime that are not statically declared in inputs or outputs.

inputsarray

The assets consumed as inputs.

io.kestra.core.models.assets.AssetIdentifier

id*string

Min length1

typestring

outputs

The assets produced as outputs.

io.kestra.plugin.ee.assets.Dataset

id*string

Min length1

Max length150

type*object

descriptionstring

displayNamestring

metadataobject

Default{}

namespacestring

Min length1

Max length150

io.kestra.plugin.ee.assets.File

id*string

Min length1

Max length150

type*object

descriptionstring

displayNamestring

metadataobject

Default{}

namespacestring

Min length1

Max length150

io.kestra.plugin.ee.assets.Table

id*string

Min length1

Max length150

type*object

descriptionstring

displayNamestring

metadataobject

Default{}

namespacestring

Min length1

Max length150

io.kestra.plugin.ee.assets.VM

id*string

Min length1

Max length150

type*object

descriptionstring

displayNamestring

metadataobject

Default{}

namespacestring

Min length1

Max length150

io.kestra.core.models.assets.External

id*string

Min length1

Max length150

type*object

descriptionstring

displayNamestring

metadataobject

Default{}

namespacestring

Min length1

Max length150

io.kestra.core.models.assets.Custom

id*string

Min length1

Max length150

type*string

Min length1

Custom asset type

descriptionstring

displayNamestring

metadataobject

Default{}

namespacestring

Min length1

Max length150

autoDelete booleanstring

Defaulttrue

Auto-delete

If true (default), delete the messages from the queue once they are stored in internal storage. Messages are held until the run ends, so a run that fails or is killed deletes nothing and its messages are delivered again once hidden time runs out. Set maxRecords to bound how many messages a run holds, since SQS caps in-flight messages per queue. The task ends in WARNING when some deletes fail, and fails when none succeed.

connectionAcquisitionTimeout string

DefaultPT5S

The amount of time to wait when acquiring a connection from the pool before giving up and timing out

endpointOverride string

The endpoint with which the SDK should communicate

This property allows you to use a different S3 compatible storage backend.

maxConcurrency integerstring

Default50

Maximum number of allowed concurrent requests

For HTTP/1.1 this is the same as max connections. For HTTP/2 the number of connections that will be used depends on the max streams allowed per connection. If the maximum number of concurrent requests is exceeded they may be queued in the HTTP client (see maxPendingConnectionAcquires) and can cause increased latencies. If the client is overloaded enough such that the pending connection queue fills up, subsequent requests may be rejected or time out (see connectionAcquisitionTimeout).

maxDuration string

Max duration

Keep polling until this duration elapses, even if the queue is empty, or until maxRecords is reached.

maxRecords integerstring

Max records

Stop after consuming this many messages.

region string

AWS region with which the SDK should communicate

secretKeyId string

Secret Key Id in order to connect to AWS

If no credentials are defined, we will use the default credentials provider chain to fetch credentials.

serdeType string

DefaultSTRING

Possible Values

STRINGJSON

The serializer/deserializer to use

sessionToken string

AWS session token, retrieved from an AWS token service, used for authenticating that this user has received temporary permissions to access a given resource

If no credentials are defined, we will use the default credentials provider chain to fetch credentials.

stsEndpointOverride string

The AWS STS endpoint with which the SDKClient should communicate

stsRoleArn string

AWS STS Role

The Amazon Resource Name (ARN) of the role to assume. If set the task will use the StsAssumeRoleCredentialsProvider. If no credentials are defined, we will use the default credentials provider chain to fetch credentials.

stsRoleExternalId string

AWS STS External Id

A unique identifier that might be required when you assume a role in another account. This property is only used when an stsRoleArn is defined.

stsRoleSessionDuration string

DefaultPT15M

AWS STS Session duration

The duration of the role session (default: 15 minutes, i.e., PT15M). This property is only used when an stsRoleArn is defined.

stsRoleSessionName string

AWS STS Session name

This property is only used when an stsRoleArn is defined.

visibilityTimeout integerstring

Default30

Visibility timeout

Seconds a received message stays hidden, from 0 to 43200. Default 30s. With autoDelete and maxDuration, received messages stay hidden for the rest of the run plus this value, so the run has this many seconds after maxDuration to store and delete them. Past the 43200 cap, a message comes back mid-run and the run stops receiving. With autoDelete and no maxDuration, a run stops receiving when one of its messages comes back, so it receives for about this many seconds at most.

Outputs

count integer

Consumed count

Messages read within limits.

failedDeletesCount integer

Default0

Messages not deleted

Stored messages whose delete failed. They stay in the queue and are delivered again. The task ends in WARNING when some deletes fail, and fails when none succeed.

uri string

Formaturi

Messages file URI

Internal storage URI with serialized messages.

Metrics

sqs.consume.messages counter

Unitmessages

Number of messages consumed from the SQS queue.