Redis Stream Claim

Redis Stream Claim

Certified

Reclaim abandoned deliveries of a Redis stream consumer group

Runs XAUTOCLAIM to take over entries that were delivered to a consumer and never acknowledged, for example because the worker or the execution that read them failed. The entries become pending for consumer, and are returned with their stream name, exact stream ID and field-value map exactly as ReadGroup returns them, so the same flow can process and then Ack them.

Only entries idle for at least minIdleTime are taken. Choose a value longer than the slowest legitimate handler: an entry that is still being processed looks idle to Redis, and taking it makes two workers handle it. The task cannot tell the two cases apart, which is why minIdleTime is required and has no default.

The work is bounded three ways. count is the page size of one XAUTOCLAIM, maxRecords caps how many entries one run takes, and maxPages caps how many pages one run scans. Redis may answer a page with no entry and a cursor other than 0-0, because it inspects a limited number of pending entries per call, so the task keeps following the cursor until it reaches 0-0 or a budget is spent. cursor and completed in the output tell the two apart: pass cursor as startId of a later run to continue the scan.

Taking an entry increments its delivery counter and resets its idle time. Entries that were deleted or trimmed from the stream while pending cannot be delivered; Redis 7.0 or later removes them from the pending list and they are listed in deletedIds instead of being returned as empty entries. Needs Redis 6.2 or later.

yaml
type: io.kestra.plugin.redis.stream.Claim

Take over entries that no worker acknowledged for 10 minutes, process them and acknowledge them.

yaml
id: redis_stream_recover
namespace: company.team

tasks:
  - id: claim
    type: io.kestra.plugin.redis.stream.Claim
    url: "{{ secret('REDIS_URI') }}"
    key: orders
    group: billing
    consumer: billing_recovery
    minIdleTime: PT10M
    maxRecords: 100

  - id: process
    type: io.kestra.plugin.core.log.Log
    message: "recovering {{ outputs.claim.count }} entries: {{ outputs.claim.messages }}"
    runIf: "{{ outputs.claim.count > 0 }}"

  - id: acknowledge
    type: io.kestra.plugin.redis.stream.Ack
    url: "{{ secret('REDIS_URI') }}"
    key: orders
    group: billing
    ids: "{{ outputs.claim.ids }}"
    runIf: "{{ outputs.claim.count > 0 }}"

Scan a large pending list in several runs by keeping the cursor of the previous run.

yaml
id: redis_stream_recover_cursor
namespace: company.team

inputs:
  - id: start
    type: STRING
    defaults: "0-0"

tasks:
  - id: claim
    type: io.kestra.plugin.redis.stream.Claim
    url: "{{ secret('REDIS_URI') }}"
    key: orders
    group: billing
    consumer: billing_recovery
    minIdleTime: PT10M
    startId: "{{ inputs.start }}"
    count: 50
    maxPages: 5
    fetchType: NONE

  - id: report
    type: io.kestra.plugin.core.log.Log
    message: "claimed {{ outputs.claim.count }}, scan completed: {{ outputs.claim.completed }}, resume from {{ outputs.claim.cursor }}"
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 that takes the entries over

Rendered consumer name. Claimed entries become pending for this consumer, and stay so until they are acknowledged. Use a stable name for the recovery role, not the name of the consumer that lost them.

Consumer group name

Rendered group name whose pending entries are scanned. The group must already exist.

Redis stream key

Rendered key passed to XAUTOCLAIM.

Minimum idle time before an entry can be taken

Only pending entries not delivered or claimed for at least this long are taken. Must be strictly positive and is required: choose a value longer than the slowest handler that reads from this group, otherwise an entry that is still being processed is processed a second time. Truncated to whole milliseconds.

Redis connection string

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.

Definitions
assetFailureBehaviorstring
Possible Values
IGNOREFAILWARN

Asset 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.

enableAutobooleanstring

Whether to auto-register assets referenced dynamically at runtime that are not statically declared in inputs or outputs.

inputsarray

The assets consumed as inputs.

id*string
Min length1
typestring
outputs

The assets produced as outputs.

id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*string
Min length1

Custom asset type

descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
Default100

Page size of one scan step

Defaults to 100 and maps to COUNT. Must be between 1 and 10000. Redis inspects a limited number of pending entries per call, so a page can hold fewer entries than this, or none. Lowered to the remaining maxRecords budget so a run never takes more than it can return.

DefaultFETCH
Possible Values
STOREFETCHFETCH_ONENONE

How entries are returned

Defaults to FETCH, which returns every taken entry in messages. FETCH_ONE returns a single entry in message and forces maxRecords and count to 1. STORE writes the entries as an ION file to Kestra internal storage and returns its uri. NONE returns no entry body; count, ids and the cursor are still populated.

Default10

Maximum scan steps of one run

Defaults to 10. Must be between 1 and 1000. Each step is one XAUTOCLAIM. This bounds the work when Redis keeps answering with empty pages and a non-final cursor. When it is spent before the cursor reached 0-0, completed is false and cursor is where the next run should continue.

Default100

Maximum entries taken by one run

Defaults to 100. Must be between 1 and 10000. The run stops as soon as this many entries were taken. Forced to 1 when fetchType is FETCH_ONE, so no entry is taken that the flow cannot see.

Default0-0

Stream ID to start the scan from

Defaults to 0-0, the start of the pending list. Pass the cursor output of a previous run to continue its scan. Must be 0-0 or a stream ID such as 1700000000000-0.

Whether the scan reached the end of the pending list

False when maxRecords or maxPages was spent first; cursor is where to continue.

Consumer the entries are now pending for

Number of entries taken over by this run

Where the scan stopped

0-0 when the whole pending list was scanned, otherwise the startId to give the next run.

Default0

Number of pending entries that no longer exist in the stream

Total across all scan steps. deletedIds lists the first 1000 only, so a heavily trimmed stream does not flood the output.

SubTypestring

Stream IDs that were pending but no longer exist in the stream

Entries deleted or trimmed while pending. Redis 7.0 or later removes them from the pending list, so they are not returned as entries and need no acknowledgement. Lists at most the first 1000 IDs, see deletedCount for the total. Always empty on older servers.

Consumer group that was scanned

SubTypestring

Exact stream IDs taken over by this run

In delivery order, ready to be passed to Ack. Populated for every fetchType.

Single entry taken over by this run

Only set when fetchType is FETCH_ONE; null when nothing was taken.

Definitions
fieldsobject
SubTypestring

Field-value map of the entry

Returned exactly as stored in Redis, with no serialization format applied.

idstring

Exact stream ID of the entry

Pass it to Ack to remove the entry from the group's pending list.

streamstring

Stream the entry was read from

Entries taken over by this run

Only set when fetchType is FETCH.

Definitions
fieldsobject
SubTypestring

Field-value map of the entry

Returned exactly as stored in Redis, with no serialization format applied.

idstring

Exact stream ID of the entry

Pass it to Ack to remove the entry from the group's pending list.

streamstring

Stream the entry was read from

Number of scan steps this run needed

Stream that was scanned

Formaturi

URI of the ION file holding the entries

Only set when fetchType is STORE.

Unitrecords

Number of entries taken over by XAUTOCLAIM.

More Plugins in this Category

Dataform
Dataform

Tasks that run Dataform data transformations through the Dataform CLI.

Data

1 tasks

6 blueprints

Huawei
Huawei

Huawei plugin for Kestra

CloudDataInfrastructure

45 tasks

6 blueprints