Alibaba Cloud Consume
CertifiedConsume messages from an MNS queue
Alibaba Cloud Consume
Consume messages from an MNS queue
Receives messages until the queue is empty, maxRecords is reached or maxDuration (20 seconds by default) elapses, stores them in an ION file in internal storage and then, by default, deletes them from the queue. Delivery is at least once: messages are only deleted after the file is stored.
type: io.kestra.plugin.alibaba.mns.ConsumeExamples
Consume up to 100 JSON messages
id: mns_consume
namespace: company.team
tasks:
- id: consume
type: io.kestra.plugin.alibaba.mns.Consume
accessKeyId: "{{ secret('ALIBABA_ACCESS_KEY_ID') }}"
accessKeySecret: "{{ secret('ALIBABA_ACCESS_KEY_SECRET') }}"
region: cn-hangzhou
accountId: "{{ secret('ALIBABA_ACCOUNT_ID') }}"
queue: orders
serdeType: JSON
maxRecords: 100
maxDuration: PT20S
Properties
accessKeyId *string
AccessKey ID
The AccessKey ID used to authenticate against Alibaba Cloud.
accessKeySecret *string
AccessKey secret
The AccessKey secret paired with the AccessKey ID.
queue *string
Queue
The name of the MNS queue.
accountId string
Account ID
The Alibaba Cloud account ID, used to derive the default endpoint https://<accountId>.mns.<region>.aliyuncs.com. Not needed when endpointOverride is set.
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.
io.kestra.core.models.assets.AssetsDeclaration
IGNOREFAILWARNAsset 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.
Whether to auto-register assets referenced dynamically at runtime that are not statically declared in inputs or outputs.
The assets consumed as inputs.
io.kestra.core.models.assets.AssetIdentifier
1The assets produced as outputs.
io.kestra.plugin.ee.assets.Dataset
1150{}1150io.kestra.plugin.ee.assets.File
1150{}1150io.kestra.plugin.ee.assets.Table
1150{}1150io.kestra.plugin.ee.assets.VM
1150{}1150io.kestra.core.models.assets.External
1150{}1150io.kestra.core.models.assets.Custom
11501Custom asset type
{}1150autoDelete booleanstring
trueAuto-delete
Delete messages from the queue once the ION file is stored. When false, they become visible again after the queue's visibility timeout.
endpointOverride string
Endpoint override
Custom http or https endpoint URL. When set, it takes precedence over the endpoint derived from region.
maxDuration string
PT20SMax duration
Stop after this duration elapses, 20 seconds by default. Keep it below the queue's visibility timeout (30 seconds by default): messages are deleted at the end of the run, and a message whose receipt handle expired in the meantime is delivered again.
maxRecords integerstring
1000Max records
Stop after consuming this many messages, at least 1, 1000 by default.
region string
Region
The Alibaba Cloud region ID, for example cn-hangzhou. The service endpoint is derived from it unless endpointOverride is set.
securityToken string
Security token
Optional Security Token Service (STS) token, required when using temporary credentials.
serdeType string
STRINGSTRINGJSONSerde type
STRING keeps message bodies as text, JSON parses them. A body that is not valid JSON is kept as a string and logged.
Outputs
count integer
Number of messages consumed
uri string
uriURI of the ION file holding the messages in internal storage, unset when no message was received
Metrics
mns.consume.messages counter
messagesNumber of messages consumed from the MNS queue.
