XCLAIM

Syntax text Syntax diagram API methods

XCLAIM key group consumer min-idle-time id [id ...] [IDLE ms]
  [TIME unix-time-milliseconds] [RETRYCOUNT count] [FORCE] [JUSTID]
  [LASTID lastid]

Client:

xclaim(
    name: KeyT,
    groupname: GroupT,
    consumername: ConsumerT,
    min_idle_time: int,
    message_ids: Union[List[StreamIdT], Tuple[StreamIdT]],
    idle: Optional[int] = None,
    time: Optional[int] = None,
    retrycount: Optional[int] = None,
    force: bool = False,
    justid: bool = False
) → ResponseT

XCLAIM(
    key: RedisArgument,
    group: RedisArgument,
    consumer: RedisArgument,
    minIdleTime: number,
    id: RedisVariadicArgument,
    options?: XClaimOptions
) → Any

xclaim(
    key: byte[],
    group: byte[],
    consumerName: byte[],
    params: long minIdleTime XClaimParams,
    ids: byte[]...
) → List<byte[]>

xclaimJustId(
    key: byte[],
    group: byte[],
    consumerName: byte[],
    params: long minIdleTime XClaimParams,
    ids: byte[]...
) → List<byte[]>

xclaim(
    key: String,
    group: String,
    consumerName: String,
    params: long minIdleTime XClaimParams,
    ids: StreamEntryID...
) → List<StreamEntry>

xclaimJustId(
    key: String,
    group: String,
    minIdleTime: String consumerName long,
    params: XClaimParams,
    ids: StreamEntryID...
) → List<StreamEntryID>

xclaim(
    key: String,
    group: String,
    consumerName: String,
    params: long minIdleTime XClaimParams,
    ids: StreamEntryID...
) → List<StreamEntry>

xclaim(
    key: K,  // the stream key.
    consumer: Consumer<K>,  // consumer identified by group name and consumer key.
    minIdleTime: long,
    messageIds: String...  // message Id's to claim.
) → List<StreamMessage<K, V>>  // simple-reply the StreamMessage.

xclaim(
    key: K,  // the stream key.
    consumer: Consumer<K>,  // consumer identified by group name and consumer key.
    args: XClaimArgs,
    messageIds: String...  // message Id's to claim.
) → List<StreamMessage<K, V>>  // simple-reply the StreamMessage.

xclaim(
    key: K,  // the stream key.
    consumer: Consumer<K>,  // consumer identified by group name and consumer key.
    minIdleTime: long,
    messageIds: String...  // message Id's to claim.
) → RedisFuture<List<StreamMessage<K, V>>>  // simple-reply the StreamMessage.

xclaim(
    key: K,  // the stream key.
    consumer: Consumer<K>,  // consumer identified by group name and consumer key.
    args: XClaimArgs,
    messageIds: String...  // message Id's to claim.
) → RedisFuture<List<StreamMessage<K, V>>>  // simple-reply the StreamMessage.

xclaim(
    key: K,  // the stream key.
    consumer: Consumer<K>,  // consumer identified by group name and consumer key.
    minIdleTime: long,
    messageIds: String...  // message Id's to claim.
) → Flux<StreamMessage<K, V>>  // simple-reply the StreamMessage.

xclaim(
    key: K,  // the stream key.
    consumer: Consumer<K>,  // consumer identified by group name and consumer key.
    args: XClaimArgs,
    messageIds: String...  // message Id's to claim.
) → Flux<StreamMessage<K, V>>  // simple-reply the StreamMessage.

XClaim(
    ctx: context.Context,
    a: *XClaimArgs
) → *XMessageSliceCmd

XClaimJustID(
    ctx: context.Context,
    a: *XClaimArgs
) → *StringSliceCmd

StreamClaim(
    key: RedisKey,  // The key of the stream.
    consumerGroup: RedisValue,  // The consumer group.
    claimingConsumer: RedisValue,  // The consumer claiming the given message(s).
    minIdleTimeInMs: long,  // The minimum message idle time to allow the reassignment of the message(s).
    messageIds: RedisValue[],  // The IDs of the messages to claim for the given consumer.
    flags: CommandFlags  // The flags to use for this operation.
) → StreamEntry[]  // The messages successfully claimed by the given consumer.

StreamClaim(
    key: RedisKey,  // The key of the stream.
    consumerGroup: RedisValue,  // The consumer group.
    claimingConsumer: RedisValue,  // The consumer claiming the given message(s).
    minIdleTimeInMs: long,  // The minimum message idle time to allow the reassignment of the message(s).
    messageIds: RedisValue[],  // The IDs of the messages to claim for the given consumer.
    flags: CommandFlags  // The flags to use for this operation.
) → StreamEntry[]  // The messages successfully claimed by the given consumer.

StreamClaim(
    key: RedisKey,  // The key of the stream.
    consumerGroup: RedisValue,  // The consumer group.
    claimingConsumer: RedisValue,  // The consumer claiming the given message(s).
    minIdleTimeInMs: long,  // The minimum message idle time to allow the reassignment of the message(s).
    messageIds: RedisValue[],  // The IDs of the messages to claim for the given consumer.
    flags: CommandFlags  // The flags to use for this operation.
) → StreamEntry[]  // The messages successfully claimed by the given consumer.

StreamClaim(
    key: RedisKey,  // The key of the stream.
    consumerGroup: RedisValue,  // The consumer group.
    claimingConsumer: RedisValue,  // The consumer claiming the given message(s).
    minIdleTimeInMs: long,  // The minimum message idle time to allow the reassignment of the message(s).
    messageIds: RedisValue[],  // The IDs of the messages to claim for the given consumer.
    flags: CommandFlags  // The flags to use for this operation.
) → StreamEntry[]  // The messages successfully claimed by the given consumer.

xclaim(
    $key: string,
    $group: string,
    $consumer: string,
    $minIdleTime: int,
    $ids: string|array,
    ?int $idle = null: Any,
    ?int $time = null: Any,
    ?int $retryCount = null: Any,
    bool $force = false: Any,
    bool $justId = false: Any,
    ?string $lastId = null: Any
) → array

xclaim(
    key: K,
    group: G,
    consumer: C,
    min_idle_time: MIT,
    ids: &'a [ID]
) → (streams::StreamClaimReply)

xclaim_options(
    key: K,
    group: G,
    consumer: C,
    min_idle_time: MIT,
    ids: &'a [ID],
    options: streams
) → Generic

xclaim(
    key: K,
    group: G,
    consumer: C,
    min_idle_time: MIT,
    ids: &'a [ID]
) → (streams::StreamClaimReply)

xclaim_options(
    key: K,
    group: G,
    consumer: C,
    min_idle_time: MIT,
    ids: &'a [ID],
    options: streams
) → Generic

Available since:: Redis Open Source 5.0.0
Time complexity:: O(log N) with N being the number of messages in the PEL of the consumer group.
ACL categories:: @write, @stream, @fast,
Compatibility:: Redis Software and Redis Cloud compatibility

In the context of a stream consumer group, this command changes the ownership of a pending message, so that the new owner is the consumer specified as the command argument. Normally this is what happens:

There is a stream with an associated consumer group.
Some consumer A reads a message via XREADGROUP from a stream, in the context of that consumer group.
As a side effect a pending message entry is created in the Pending Entries List (PEL) of the consumer group: it means the message was delivered to a given consumer, but it was not yet acknowledged via XACK.
Then suddenly that consumer fails forever.
Other consumers may inspect the list of pending messages, that are stale for quite some time, using the XPENDING command. In order to continue processing such messages, they use XCLAIM to acquire the ownership of the message and continue. Consumers can also use the XAUTOCLAIM command to automatically scan and claim stale pending messages.

This dynamic is clearly explained in the Stream intro documentation.

Note that the message is claimed only if its idle time is greater than the minimum idle time we specify when calling XCLAIM. Because as a side effect XCLAIM will also reset the idle time (since this is a new attempt at processing the message), two consumers trying to claim a message at the same time will never both succeed: only one will successfully claim the message. This avoids that we process a given message multiple times in a trivial way (yet multiple processing is possible and unavoidable in the general case).

Messages that have been released back to the group using XNACK are immediately claimable since their delivery time is set to 0, satisfying any minimum idle time requirement.

Moreover, as a side effect, XCLAIM will increment the count of attempted deliveries of the message unless the JUSTID option has been specified (which only delivers the message ID, not the message itself). In this way messages that cannot be processed for some reason, for instance because the consumers crash attempting to process them, will start to have a larger counter and can be detected inside the system.

XCLAIM will not claim a message in the following cases:

The message doesn't exist in the group PEL (i.e. it was never read by any consumer)
The message exists in the group PEL but not in the stream itself (i.e. the message was read but never acknowledged, and then was deleted from the stream, either by trimming or by XDEL)

In both cases the reply will not contain a corresponding entry to that message (i.e. the length of the reply array may be smaller than the number of IDs provided to XCLAIM). In the latter case, the message will also be deleted from the PEL in which it was found. This feature was introduced in Redis 7.0.

Required arguments

key

The stream key.

group

The consumer group name.

consumer

The name of the consumer that will own the claimed messages.

min-idle-time

Claim only messages that have been idle for at least this long, in milliseconds.

id [id ...]

One or more message IDs to claim.

Optional arguments

Most of the following options are mainly for internal use, to transfer the effects of XCLAIM to the AOF file and replicas, and are unlikely to be useful to normal users.

IDLE ms

Set the idle time (last time it was delivered) of the message. If IDLE is not specified, an idle time of 0 is assumed, that is, the time count is reset because the message now has a new owner trying to process it.

TIME unix-time-milliseconds

The same as IDLE but sets the idle time to a specific Unix time (in milliseconds) instead of a relative amount of milliseconds. This is useful in order to rewrite the AOF file generating XCLAIM commands.

RETRYCOUNT count

Set the retry counter to the specified value. If not set, XCLAIM increments the retry counter every time a message is delivered again.

FORCE

Create the pending message entry in the PEL even if the specified IDs are not already in the PEL assigned to a different client. However, the message must exist in the stream, otherwise the IDs of non-existing messages are ignored.

JUSTID

Return just an array of IDs of messages successfully claimed, without returning the actual messages. Using this option means the retry counter is not incremented.

LASTID lastid

Update the consumer group's last-delivered ID to the given ID.

Examples

> XCLAIM mystream mygroup Alice 3600000 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "message"
      2) "orange"

In the previous example, you claim the message with ID 1526569498055-0 and assign it to the consumer Alice, but only if neither the original consumer nor another consumer has acknowledged or claimed it for at least one hour.

Redis Software and Redis Cloud compatibility

Redis Software	Redis Cloud	Notes
✅ Standard ✅ Active-Active	✅ Standard ✅ Active-Active

Return information

RESP2 RESP3

Any of the following:

Array reply: when the JUSTID option is specified, an array of IDs of messages successfully claimed.
Array reply: an array of stream entries, each of which contains an array of two elements, the entry ID and the entry data itself.