Redis Stream ReadGroup
CertifiedRead entries from a Redis stream consumer group
Redis Stream ReadGroup
Read entries from a Redis stream consumer group
Runs one bounded XREADGROUP for the given consumer and returns each entry with its stream name, its exact stream ID and its field-value map. Field names and values are returned exactly as Redis stores them, with no serialization format applied, so an ID can be fed straight back into Ack.
The read is bounded on both axes. count caps how many entries a single call delivers, and blockDuration caps how long the call waits when the group has nothing to deliver; leaving blockDuration unset makes the call return immediately, and it cannot exceed one hour. XREADGROUP BLOCK 0, which waits forever, is deliberately not reachable. Killing the task closes the connection, so a waiting read stops instead of holding a worker thread.
Delivery is at-least-once. Every entry this task returns is added to the group's Pending Entries List for this consumer and stays there until Ack removes it, so a flow that fails after the read leaves the entry pending and a later run can pick it up again. The consumer name is therefore part of the contract: entries delivered to one consumer name remain owned by that name. Use a stable name per worker role, keep processing idempotent because a crash between a business write and the acknowledgement replays the entry, and set offset to 0 to re-read this consumer's own unacknowledged entries instead of new ones.
type: io.kestra.plugin.redis.stream.ReadGroupExamples
Read, process and acknowledge entries from a stream consumer group.
id: redis_stream_consume
namespace: company.team
tasks:
- id: create_group
type: io.kestra.plugin.redis.stream.GroupCreate
url: "{{ secret('REDIS_URI') }}"
key: orders
group: billing
startId: "0"
- id: read
type: io.kestra.plugin.redis.stream.ReadGroup
url: "{{ secret('REDIS_URI') }}"
key: orders
group: billing
consumer: billing_worker_1
count: 10
blockDuration: PT5S
- id: process
type: io.kestra.plugin.core.log.Log
message: "handling {{ outputs.read.count }} entries: {{ outputs.read.messages }}"
- id: acknowledge
type: io.kestra.plugin.redis.stream.Ack
url: "{{ secret('REDIS_URI') }}"
key: orders
group: billing
ids: "{{ outputs.read.ids }}"
runIf: "{{ outputs.read.count > 0 }}"
Recover this consumer's own entries that were read but never acknowledged.
id: redis_stream_recover_own
namespace: company.team
tasks:
- id: read_pending
type: io.kestra.plugin.redis.stream.ReadGroup
url: "{{ secret('REDIS_URI') }}"
key: orders
group: billing
consumer: billing_worker_1
offset: "0"
count: 100
Store a batch in Kestra internal storage instead of returning it inline.
id: redis_stream_store
namespace: company.team
tasks:
- id: read
type: io.kestra.plugin.redis.stream.ReadGroup
url: "{{ secret('REDIS_URI') }}"
key: orders
group: billing
consumer: billing_worker_1
fetchType: STORE
count: 500
Properties
consumer *string
Consumer name inside the group
Rendered consumer name. Redis tracks pending entries per consumer, so entries delivered under this name stay owned by it until they are acknowledged. Use a stable name per worker role rather than a value that changes on every execution, otherwise unacknowledged entries are stranded under a name nothing reads again.
group *string
Consumer group name
Rendered group name; the group must already exist, so create it with GroupCreate first. Reading from a missing group fails with NOGROUP.
key *string
Redis stream key
Rendered key passed to XREADGROUP.
url *string
Redis connection string
assets
Assets this task consumes as inputs or produces as outputs, for lineage tracking and the asset graph (Enterprise Edition). A flow declaring this property on a task is rejected in the open-source edition.
io.kestra.core.models.assets.AssetsDeclaration
IGNOREFAILWARNAsset failure behavior
Behavior applied to the task state when a declared asset fails to render, emit, or be persisted (e.g. a lock conflict): FAIL escalates it to FAILED, WARN (default) warns it if it would otherwise succeed, IGNORE leaves the state untouched.
Whether to auto-register assets referenced dynamically at runtime that are not statically declared in inputs or outputs.
The assets consumed as inputs.
io.kestra.core.models.assets.AssetIdentifier
1The assets produced as outputs.
io.kestra.plugin.ee.assets.Dataset
1150{}1150io.kestra.plugin.ee.assets.File
1150{}1150io.kestra.plugin.ee.assets.Table
1150{}1150io.kestra.plugin.ee.assets.VM
1150{}1150io.kestra.core.models.assets.External
1150{}1150io.kestra.core.models.assets.Custom
11501Custom asset type
{}1150blockDuration string
How long to wait for entries
Maps to BLOCK. Unset by default, which makes the read return immediately with whatever is available. When set it must be strictly positive and at most 1 hour (PT1H); BLOCK 0, which waits forever, is not supported because the read has to stay bounded. The client command timeout is raised above this value for the call. The task-level timeout can cap the wait further.
count integerstring
100Maximum entries per read
Defaults to 100 and maps to COUNT. Must be between 1 and 10000. Forced to 1 when fetchType is FETCH_ONE, so the task never makes Redis mark entries as delivered that the flow cannot see or acknowledge.
fetchType string
FETCHSTOREFETCHFETCH_ONENONEHow entries are returned
Defaults to FETCH, which returns every entry of the batch in messages. FETCH_ONE returns a single entry in message and forces count to 1. STORE writes the batch as an ION file to Kestra internal storage and returns its uri. NONE returns no entry body; count and ids are still populated, so the batch can be acknowledged or counted without carrying the payload.
noAck booleanstring
falseSkip the pending list for this read
Defaults to false. When true the read adds NOACK, so Redis never records the entries as pending and Ack has nothing to remove. This downgrades delivery to at-most-once: an entry lost by a failing flow is not redelivered. Leave it false whenever the work must not be lost.
offset string
>Stream ID to read from
Defaults to >, which delivers entries never handed to any consumer of the group and adds them to this consumer's pending list. Use 0 to re-read this consumer's own already delivered but unacknowledged entries, oldest first, or any explicit stream ID to re-read its pending entries after that ID. Only > delivers new entries.
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.
Metrics
read.entries.count counter
recordsNumber of entries delivered by XREADGROUP.
