AMQP Amqp1Trigger

AMQP Amqp1Trigger
Certified
Poll an AMQP 1.0 address into batch executions

Polls the address every 60 seconds by default, consumes like io.kestra.plugin.amqp.Amqp1Consume until maxRecords, maxDuration or pollDuration is reached, and launches one execution per non-empty batch with payloads stored at trigger.uri. Set at least one of maxRecords or maxDuration.

yaml
type: io.kestra.plugin.amqp.Amqp1Trigger

Examples

Start an execution for each batch of up to 100 messages received on an Azure Service Bus queue.

yaml
id: azure_service_bus_trigger
namespace: company.team

tasks:
  - id: log
    type: io.kestra.plugin.core.log.Log
    message: "Received {{ trigger.count }} messages stored at {{ trigger.uri }}"

triggers:
  - id: trigger
    type: io.kestra.plugin.amqp.Amqp1Trigger
    host: my-namespace.servicebus.windows.net
    ssl: true
    username: RootManageSharedAccessKey
    password: "{{ secret('SERVICE_BUS_SAS_KEY') }}"
    address: orders
    maxRecords: 100

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

DefaultSTRING

Possible Values

STRINGJSON

Payload serde format

Controls how message bodies are read; use STRING for raw text or JSON for structured data. Defaults to STRING.

allowConcurrent boolean

Defaultfalse

Specifies whether a trigger is allowed to start a new execution even if a previous run is still in progress.

autoAck booleanstring

Defaultfalse

Automatic 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

DefaultPT30S

Connection 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.

interval string

DefaultPT1M

Formatduration

Interval between polling.

The interval between 2 different polls of schedule, this can avoid to overload the remote system with too many calls. For most of the triggers that depend on external systems, a minimal interval must be at least PT30S. See ISO_8601 Durations for more information of available interval values.

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

DefaultPT5S

Poll 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

Defaultfalse

Use 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') }}.

stopAfter array

SubTypestring

Possible Values

CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTED

List of execution states after which a trigger should be stopped (a.k.a. disabled).

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.

when string

Defaulttrue

A condition that determines whether the trigger should run.

A Pebble expression evaluated at trigger time. The trigger fires only when the expression evaluates to a truthy value (true, a non-empty string, a non-zero number). Use this to gate trigger execution on dynamic runtime values such as execution labels, flow variables, or environment conditions.

Outputs

count integer

Total messages consumed

Count of messages consumed before a stop condition was reached.

uri string

Formaturi

URI of file storing consumed messages

Internal storage path to the ION-serialized batch, returned as taskrun.outputs.uri or trigger.uri.