AMQP Amqp1Consume
CertifiedConsume AMQP 1.0 messages until a stop condition
AMQP Amqp1Consume
Consume AMQP 1.0 messages until a stop condition
Consumes from a queue or subscription over AMQP 1.0, writing messages to internal storage and returning their URI; requires maxDuration or maxRecords, and also stops when no message arrives within pollDuration. Messages are accepted once the batch is durably stored, and released back to the broker on failure or kill. Works with Azure Service Bus, ActiveMQ Artemis, RabbitMQ 4 and other AMQP 1.0 brokers; use io.kestra.plugin.amqp.Consume for AMQP 0.9.1.
type: io.kestra.plugin.amqp.Amqp1ConsumeExamples
Consume up to 1000 messages from an ActiveMQ Artemis queue.
id: amqp1_consume
namespace: company.team
tasks:
- id: consume
type: io.kestra.plugin.amqp.Amqp1Consume
host: localhost
port: 5672
username: artemis
password: "{{ secret('AMQP_PASSWORD') }}"
address: orders
maxRecords: 1000
Consume JSON messages from an Azure Service Bus topic subscription for at most 30 seconds.
id: azure_service_bus_consume
namespace: company.team
tasks:
- id: consume
type: io.kestra.plugin.amqp.Amqp1Consume
host: my-namespace.servicebus.windows.net
ssl: true
username: RootManageSharedAccessKey
password: "{{ secret('SERVICE_BUS_SAS_KEY') }}"
address: orders/Subscriptions/kestra
serdeType: JSON
maxDuration: PT30S
Properties
address *string
Source address
Address to consume from, in the broker's format: the queue name for ActiveMQ Artemis and Azure Service Bus (<topic>/Subscriptions/<subscription> for a Service Bus topic subscription), or /queues/<queue> for RabbitMQ 4.
host *string
Broker host
Hostname or IP of the AMQP 1.0 broker, e.g. my-namespace.servicebus.windows.net for Azure Service Bus.
serdeType *string
STRINGSTRINGJSONPayload serde format
Controls how message bodies are read; use STRING for raw text or JSON for structured data. Defaults to STRING.
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
{}1150autoAck booleanstring
falseAutomatic acknowledgment
When true, each message is accepted as soon as it is received; it is not redelivered if the task is killed mid-batch.
When false, every message of the batch is accepted once the batch is durably stored, and released back to the broker
if the task fails or is killed before that point. On Azure Service Bus (peek-lock), keep the batch shorter than the
queue's lock duration. Defaults to false.
connectionTimeout string
PT30SConnection timeout
Maximum time to open or close the connection and its links, and to wait for the broker to settle each published message. Defaults to PT30S.
maxDuration string
Maximum duration
Time after which consumption stops. Required when maxRecords is not set.
maxRecords integerstring
Maximum records
Number of messages after which consumption stops; credit is never granted beyond it, so the broker keeps the remaining messages. Required when maxDuration is not set.
password string
Password
SASL PLAIN password; for Azure Service Bus use the shared access key.
pollDuration string
PT5SPoll duration
Maximum time to wait for the next message; consumption stops early when none arrives within it, e.g. once the queue is drained. Defaults to PT5S.
port integerstring
Broker port
TCP port of the AMQP 1.0 listener; defaults to 5672, or 5671 when ssl is enabled.
ssl booleanstring
falseUse TLS (AMQPS)
Encrypts the connection with TLS. The broker certificate and hostname are always verified, against the JVM's trusted certificate authorities unless sslCaCertificate is set. Required by Azure Service Bus. Defaults to false.
sslCaCertificate string
CA certificate for TLS
PEM-encoded certificate (or chain) of the authority that signed the broker certificate, for brokers using a private or self-signed CA. Only used when ssl is true; when set, only this CA is trusted. Example: {{ secret('AMQP_CA_CERT') }}.
username string
Username
SASL PLAIN username; for Azure Service Bus use the shared access policy name (e.g. RootManageSharedAccessKey). SASL ANONYMOUS is used when not set.
Outputs
count integer
Total messages consumed
Count of messages consumed before a stop condition was reached.
uri string
uriURI of file storing consumed messages
Internal storage path to the ION-serialized batch, returned as taskrun.outputs.uri or trigger.uri.
Metrics
consumed.records counter
recordsThe total number of records consumed from the AMQP 1.0 address.
