Redis Stream Trigger
CertifiedStart an execution for new entries of a Redis stream consumer group
Redis Stream Trigger
Start an execution for new entries of a Redis stream consumer group
Periodically reads one batch of new entries with XREADGROUP as consumer inside group, using the same contract as ReadGroup, and starts one execution when the batch is not empty. The execution gets the stream name, the exact stream IDs and the field-value maps of the entries as trigger.ids, trigger.messages or trigger.uri, depending on fetchType. The group must already exist: create it once with GroupCreate.
The trigger never acknowledges anything. Delivery is at-least-once: the entries of a batch are pending for consumer as soon as they are read, and stay so until the flow runs Ack with {{ trigger.ids }} after the business step. When the execution fails before that, the entries are not read again by this trigger, because it only asks for new entries. Recover them with a flow that runs Claim. Handlers must be idempotent, because a failure between a business write and the acknowledgement replays the entry.
Every evaluation is a single read that returns immediately, so a quiet stream costs one command per interval and never holds a scheduler thread. Use a stable consumer name for the trigger and a different one for any other reader of the same group. Killing the trigger closes the connection of a read in flight and starts no execution.
type: io.kestra.plugin.redis.stream.TriggerExamples
Start an execution for each batch of new orders, process it and acknowledge it.
id: redis_stream_listen
namespace: company.team
tasks:
- id: process
type: io.kestra.plugin.core.log.Log
message: "handling {{ trigger.count }} entries: {{ trigger.messages }}"
- id: acknowledge
type: io.kestra.plugin.redis.stream.Ack
url: "{{ secret('REDIS_URI') }}"
key: orders
group: billing
ids: "{{ trigger.ids }}"
triggers:
- id: watch
type: io.kestra.plugin.redis.stream.Trigger
url: "{{ secret('REDIS_URI') }}"
key: orders
group: billing
consumer: billing_trigger
count: 50
interval: PT30S
Keep large batches out of the execution by storing them, and still acknowledge by ID.
id: redis_stream_listen_store
namespace: company.team
tasks:
- id: acknowledge
type: io.kestra.plugin.redis.stream.Ack
url: "{{ secret('REDIS_URI') }}"
key: orders
group: billing
ids: "{{ trigger.ids }}"
triggers:
- id: watch
type: io.kestra.plugin.redis.stream.Trigger
url: "{{ secret('REDIS_URI') }}"
key: orders
group: billing
consumer: billing_trigger
fetchType: STORE
Installation
Our standard Docker images already include this plugin. If you use a -slim image or another setup, install it with the following command:
kestractl plugins get io.kestra.plugin:plugin-redis --compatible-for $KESTRA_VERSION For kestractl, use your Kestra version with --compatible-for, or specify an exact plugin version.
For Kestra CLI, replace LATEST with a specific version or version range to pin the plugin version.
Properties
consumer *string
Consumer name inside the group
Rendered consumer name. Entries read by the trigger are pending for this consumer until they are acknowledged. Use a stable name that no other reader of the group uses.
group *string
Consumer group name
Rendered group name; the group must already exist, so create it with GroupCreate first.
key *string
Redis stream key
Rendered key passed to XREADGROUP.
url *string
Redis connection string
allowConcurrent boolean
falseSpecifies whether a trigger is allowed to start a new execution even if a previous run is still in progress.
count integerstring
100Maximum entries per evaluation
Defaults to 100 and maps to COUNT. Must be between 1 and 1000. Forced to 1 when fetchType is FETCH_ONE, so no entry is marked as delivered without its ID reaching the execution.
fetchType string
FETCHSTOREFETCHFETCH_ONENONEHow entries are passed to the execution
Defaults to FETCH, which puts every entry of the batch in trigger.messages. FETCH_ONE passes a single entry in trigger.message. STORE writes the batch as an ION file and passes its trigger.uri, which keeps large payloads out of the execution. NONE passes no entry body. trigger.ids and trigger.count are always set, so the batch can be acknowledged in every mode.
interval string
PT1MdurationInterval 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.
stopAfter array
CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTEDList of execution states after which a trigger should be stopped (a.k.a. disabled).
when string
trueA 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
consumer string
Consumer name the entries are now pending for
count integer
Number of entries delivered by this read
group string
Consumer group that was read
ids array
Exact stream IDs delivered by this read
In delivery order, ready to be passed to Ack. Populated for every fetchType.
message
Single entry delivered by this read
Only set when fetchType is FETCH_ONE; null when the read returned nothing.
io.kestra.plugin.redis.stream.ReadGroup-Message
Field-value map of the entry
Returned exactly as stored in Redis, with no serialization format applied.
Exact stream ID of the entry
Pass it to Ack to remove the entry from the group's pending list.
Stream the entry was read from
messages array
Entries delivered by this read
Only set when fetchType is FETCH.
io.kestra.plugin.redis.stream.ReadGroup-Message
Field-value map of the entry
Returned exactly as stored in Redis, with no serialization format applied.
Exact stream ID of the entry
Pass it to Ack to remove the entry from the group's pending list.
Stream the entry was read from
stream string
Stream that was read
uri string
uriURI of the ION file holding the entries
Only set when fetchType is STORE.
