AMQP Trigger

AMQP Trigger
Certified
Poll AMQP queue into batch executions

Polls the queue every 60 seconds by default, consumes until maxRecords or maxDuration, and launches one execution per batch with payloads stored at trigger.uri. Set at least one stop condition; deprecated url is kept for backward compatibility—use host/port/virtualHost instead.

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

Examples

yaml
id: amqp_trigger
namespace: company.team

tasks:
  - id: trigger
    type: io.kestra.plugin.amqp.Trigger
    host: localhost
    port: 5672
    username: guest
    password: "{{ secret('AMQP_PASSWORD') }}"
    virtualHost: /my_vhost
    maxRecords: 2
    queue: amqpTrigger.queue

Properties

host *string

Broker host

Hostname or IP of the RabbitMQ broker; required unless using the deprecated url.

queue *string

Queue name to consume

AMQP queue to read from; required and must already exist.

serdeType *string

DefaultSTRING

Possible Values

STRINGJSON

Payload serde format

Controls how message bodies are read and written; 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, the broker acknowledges messages as soon as they are delivered; they are not requeued if the trigger is killed mid-batch. When false, the trigger sends a single bulk acknowledgment for the whole batch once it is durably stored, and NACKs a message on processing failure; if killed before that point, the broker requeues every unacknowledged message.

consumerTag string

DefaultKestra

Consumer tag

Client-supplied consumer tag used for tracing and cancellations; defaults to Kestra in tasks and triggers.

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

Soft cap on run time; checked roughly every 100 ms so actual runtime can slightly exceed this value. Required when maxRecords is not set.

maxRecords integerstring

Maximum records

Soft cap on messages consumed before stopping; evaluated after each ACKed message. Required when maxDuration is not set.

password string

Password

Password for the connection; required when the broker enforces authentication.

port string

Default5672

Broker port

TCP port for AMQP connections; defaults to 5672.

stopAfter array

SubTypestring

Possible Values

CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTED

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

username string

Username

Username for the connection; uses broker default (often guest) when not set.

virtualHost string

Default/

Virtual host

Broker virtual host path; defaults to /.

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 the stop condition was reached; acknowledged in a single batch once this output is durably stored.

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.