Consume

Consume messages from an AMQP queue.

Requires maxDuration or maxRecords.

yaml
type: "io.kestra.plugin.amqp.Consume"

Examples

yaml
id: amqp_consume
namespace: company.team

tasks:
  - id: consume
    type: io.kestra.plugin.amqp.Consume
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtualHost: /my_vhost
    queue: kestramqp.queue
    maxRecords: 1000

Properties

host *string

The broker host

queue *string

The queue to pull messages from

serdeType *string

Default STRING

Possible Values

STRINGJSON

Serialization format

Defines how message payloads are serialized or deserialized. Use STRING for plain text or JSON for structured data.

consumerTag string

Default Kestra

A client-generated consumer tag to establish context

maxDuration string

Format duration

Maximum duration

The maximum duration the consumer will run before stopping. This is a soft limit evaluated approximately every 100 milliseconds, so the actual duration may slightly exceed this value.

maxRecords integerstring

Maximum number of records

The maximum number of messages to consume before stopping. This is a soft limit evaluated after each message.

password string

The broker password

port string

Default 5672

The broker port

username string

The broker username

virtualHost string

Default /

The broker virtual host

Outputs

count integer

Number of rows consumed

uri string

Format uri

File URI containing consumed messages

Metrics

consumed.records counter

The total number of records consumed from the AMQP queue.

​Consume

Consume