NATS Trigger

NATS Trigger
Certified
Trigger on polled NATS messages

Polls a JetStream subject on a schedule (default every 60s) and starts one execution per batch. Defaults: deliverPolicy=All, pollDuration=PT2S, batchSize=10. Use the realtime io.kestra.plugin.nats.RealtimeTrigger to emit one execution per message.

yaml
type: io.kestra.plugin.nats.core.Trigger

Examples

Subscribe to a NATS subject, getting every message from the beginning of the subject on first trigger execution.

yaml
id: nats
namespace: company.team

tasks:
  - id: log
    type: io.kestra.plugin.core.log.Log
    message: "{{ trigger.data }}"

triggers:
  - id: watch
    type: io.kestra.plugin.nats.core.Trigger
    url: nats://localhost:4222
    username: nats_user
    password: "{{ secret('NATS_PASSWORD') }}"
    subject: kestra.trigger
    durableId: natsTrigger
    deliverPolicy: All
    maxRecords: 1

Properties

subject *string

Min length1

Subject to subscribe to

url *string

Min length1

URL to connect to NATS server

The format is (nats://)server_url: port. You can also provide a connection token like so: nats://token@server_url: port

allowConcurrent boolean

Defaultfalse

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

batchSize integer

Default10

Minimum>= 1

Messages are fetched by batch of given size

creds string

Credentials files authentification

deliverPolicy string

DefaultAll

Possible Values

AllLastNewByStartSequenceByStartTimeLastPerSubject

The point in the stream to receive messages from

Possible settings are:

All: The default policy. The consumer will start receiving from the earliest available message.
Last: When first consuming messages, the consumer will start receiving messages with the last message added to the stream, or the last message in the stream that matches the consumer's filter subject if defined.
New: When first consuming messages, the consumer will only start receiving messages that were created after the consumer was created.
ByStartSequence: When first consuming messages, start at the first message having the sequence number or the next one available.
ByStartTime: When first consuming messages, start with messages on or after this time. The consumer is required to specify since which defines this start time.
LastPerSubject: When first consuming messages, start with the latest one for each filtered subject currently in the stream.

durableId string

ID used to attach the subscription to a durable one, allowing the subscription to start back from a previous position

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 polling duration

Soft wall-clock limit evaluated between fetches; polling stops once exceeded.

maxRecords integerstring

Maximum records to consume

Optional cap on total messages before stopping; implementations stop once this count is reached.

password string

Plaintext authentication password

pollDuration string

DefaultPT2S

Fetch wait duration

Max wait per fetch when no messages are available; implementations default to PT2S.

since string

Minimum message timestamp to start consumption from

By default, we consume all messages from the subjects starting from beginning of logs or depending on the current durable id position. You can also provide an arbitrary start time to get all messages since this date for a new durable id. Note that if you don't provide a durable id, you will retrieve all messages starting from this date even after subsequent usage of this task.Must be a valid iso 8601 date.

stopAfter array

SubTypestring

Possible Values

CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTED

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

token string

Token authentification

username string

Plaintext authentication username

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

messagesCount integer

Messages consumed

Total acknowledged messages during this run.

uri string

Formaturi

Output file URI

Kestra internal storage URI of the ION file containing consumed messages.