Redis Stream Ack
CertifiedAcknowledge entries read from a Redis stream consumer group
Redis Stream Ack
Acknowledge entries read from a Redis stream consumer group
Runs XACK for the given stream IDs, which removes them from the group's Pending Entries List and ends the group's responsibility for them.
Acknowledging is a separate step on purpose. Put this task after the step that actually finished the work, so a failure anywhere before it leaves the entries pending and available for a later run instead of silently dropping them. Feed it the ids output of ReadGroup, which carries the exact IDs that were delivered.
XACK is idempotent and counts only the entries it actually removed, so an ID that was already acknowledged, never pending for this group, or trimmed out of the stream is simply not counted. Running the same acknowledgement twice is therefore harmless, but it also means a short count is not an error by itself: compare count with size, or set failedOnMissing to fail when they differ.
type: io.kestra.plugin.redis.stream.AckExamples
Acknowledge the entries of a read once the business step succeeded.
id: redis_stream_ack
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
count: 10
- id: process
type: io.kestra.plugin.core.log.Log
message: "handling {{ outputs.read.count }} entries"
- 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 }}"
Acknowledge a single known entry and fail if it was not pending.
id: redis_stream_ack_strict
namespace: company.team
tasks:
- id: acknowledge
type: io.kestra.plugin.redis.stream.Ack
url: "{{ secret('REDIS_URI') }}"
key: orders
group: billing
ids:
- "1700000000000-0"
failedOnMissing: true
Properties
group *string
Consumer group name
Rendered group name whose pending list the entries are removed from. XACK on a group that does not exist acknowledges nothing rather than failing.
ids *array
Stream IDs to acknowledge
Rendered list of exact stream IDs, typically the ids output of ReadGroup. Duplicates are removed before the call. An empty list is a no-op and does not call Redis.
key *string
Redis stream key
Rendered key passed to XACK.
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
{}1150failedOnMissing booleanstring
falseFail when some IDs were not pending
Defaults to false, which reports the acknowledged count without failing; this keeps a retried acknowledgement harmless. When true, the task throws if fewer entries were removed than requested, which catches IDs that were already acknowledged, never pending for this group, or trimmed out of the stream.
Outputs
count integer
Number of entries actually removed from the pending list
As reported by XACK. Lower than size when some IDs were not pending for the group.
size integer
Number of distinct IDs that were requested
Metrics
acknowledged.entries.count counter
recordsNumber of entries removed from the pending list by XACK.
