AMQP Consume

AMQP Consume

Certified

Consume AMQP messages until a stop condition

Consumes from a queue with manual ACK/NACK, writing messages to internal storage and returning their URI; requires maxDuration or maxRecords to stop. Defaults to consumer tag Kestra and serde type STRING.

yaml
type: io.kestra.plugin.amqp.Consume
yaml
id: amqp_consume
namespace: company.team

tasks:
  - id: consume
    type: io.kestra.plugin.amqp.Consume
    host: localhost
    port: 5672
    username: guest
    password: "{{ secret('AMQP_PASSWORD') }}"
    virtualHost: /my_vhost
    queue: kestramqp.queue
    maxRecords: 1000
Properties

Broker host

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

Queue name to consume

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

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.

Defaultfalse

Automatic acknowledgment

When true, the broker acknowledges messages as soon as they are delivered. When false, the task ACKs after processing and NACKs on failure.

DefaultKestra

Consumer tag

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

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.

Maximum records

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

Password

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

Reference (ref) of the pluginDefaults to apply to this task.

Default5672

Broker port

TCP port for AMQP connections; defaults to 5672.

Username

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

Default/

Virtual host

Broker virtual host path; defaults to /.

Total messages consumed

Count of messages acknowledged before the stop condition was reached.

Formaturi

URI of file storing consumed messages

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

Unitrecords

The total number of records consumed from the AMQP queue.

More Plugins in this Category

Liquibase
Liquibase

Tasks that run Liquibase change management commands.

Data

2 tasks

OPA (Open Policy Agent)
OPA (Open Policy Agent)
Enterprise

Tasks to manage Open Policy Agent policies and decisions, including upload, evaluation, and partial evaluation.

Infrastructure

5 tasks

3 blueprints