Redis Stream GroupCreate
CertifiedCreate a consumer group on a Redis stream
Redis Stream GroupCreate
Create a consumer group on a Redis stream
Runs XGROUP CREATE so that ReadGroup and Ack have a group to work with.
startId decides what the group considers already handled and is never guessed: $ starts the group at the current end of the stream so only entries added afterwards are delivered, while 0 replays the whole retained backlog. Any explicit stream ID is also accepted, so a group can start just after a known entry.
The task is safe to run on every execution: XGROUP CREATE fails with BUSYGROUP once the group exists, and ignoreExisting (on by default) turns that into a no-op while created reports whether this run actually created the group. makeStream adds MKSTREAM, which creates an empty stream when the key does not exist yet, instead of failing on a missing key.
type: io.kestra.plugin.redis.stream.GroupCreateExamples
Create a consumer group that only receives entries added after the group was created.
id: redis_stream_group_create
namespace: company.team
tasks:
- id: create_group
type: io.kestra.plugin.redis.stream.GroupCreate
url: "{{ secret('REDIS_URI') }}"
key: orders
group: billing
startId: "$"
Create a consumer group that replays the whole retained backlog of the stream.
id: redis_stream_group_create_backlog
namespace: company.team
tasks:
- id: create_group
type: io.kestra.plugin.redis.stream.GroupCreate
url: "{{ secret('REDIS_URI') }}"
key: orders
group: billing_backfill
startId: "0"
Properties
group *string
Consumer group name
Rendered group name passed to XGROUP CREATE.
key *string
Redis stream key
Rendered key passed to XGROUP CREATE.
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
{}1150ignoreExisting booleanstring
trueSucceed when the group already exists
Defaults to true, which turns the BUSYGROUP error of a second run into a no-op and leaves the existing group untouched, including its current position and pending entries. Set to false to fail instead. The created output tells the two cases apart.
makeStream booleanstring
trueCreate the stream when the key is missing
Defaults to true, which adds MKSTREAM so an empty stream is created instead of failing on a missing key. Set to false to require the stream to exist already.
startId string
$Stream ID the group starts from
Defaults to $: the group is positioned at the current end of the stream, so it only receives entries added after this task ran and the existing backlog is treated as already handled. Use 0 to deliver the whole retained backlog instead, or any explicit stream ID (for example 1700000000000-0) to start right after that entry. Entries trimmed out of the stream are never delivered, whatever the value.
Outputs
created boolean
Whether this run created the group
False when the group already existed and ignoreExisting is true.
group string
Consumer group name
key string
Stream key the group was created on
startId string
Stream ID the group was requested to start from
