XACKDEL - Upstash Documentation

The XACKDEL command acknowledges and deletes stream entries atomically in a single operation. This is useful for consumer groups where you want to acknowledge message processing and remove the entries from the stream simultaneously.

Arguments

keybodystrrequired

The key of the stream.

groupbodystrrequired

The consumer group name.

idsbody*List[str]required

One or more stream entry IDs to acknowledge and delete.

optionbodyLiteral['KEEPREF', 'DELREF', 'ACKED', 'keepref', 'delref', 'acked']

Optional deletion behavior (case-insensitive):

KEEPREF or keepref: Keep consumer group references
DELREF or delref: Delete consumer group references
ACKED or acked: Only acknowledge messages (don't delete)

Response

List[int]required

A list of integers indicating the result for each ID in the same order as provided.

Single Entry

# Acknowledge and delete a single entryresult = redis.xackdel("mystream", "mygroup", "1638360173533-0")print(result)  # List of results for each ID

Multiple Entries

# Acknowledge and delete multiple entriesresult = redis.xackdel(    "mystream",    "mygroup",    "1638360173533-0",    "1638360173533-1",    "1638360173533-2")print(result)  # List of results for each ID

With KEEPREF Option

# Keep consumer group references when deleting entries# Useful when you want to delete entries but maintain group trackingresult = redis.xackdel(    "mystream",    "mygroup",    "1638360173533-0",    "1638360173533-1",    option="KEEPREF")print(result)  # List of results for each ID

With DELREF Option

# Delete consumer group references along with entries# Useful for complete cleanup of processed messagesresult = redis.xackdel(    "mystream",    "mygroup",    "1638360173533-0",    "1638360173533-1",    option="DELREF")print(result)  # List of results for each ID

With ACKED Option

# Only acknowledge messages without deleting them# Useful when you want to mark as processed but keep for audit/debuggingresult = redis.xackdel(    "mystream",    "mygroup",    "1638360173533-0",    option="ACKED")print(result)  # List of results for each ID# Entry remains in stream but is acknowledged

Consumer Group Processing

# Create a consumer groupredis.xgroup_create("orders", "processors", "0", mkstream=True)# Add some orders to the streamredis.xadd("orders", "*", {"order_id": "123", "status": "pending"})redis.xadd("orders", "*", {"order_id": "124", "status": "pending"})# Read messages as a consumermessages = redis.xreadgroup(    "processors", "consumer1", {"orders": ">"},    count=2)# Process messages and acknowledge + delete themfor stream_name, entries in messages:    for entry_id, fields in entries:        # Process the order...        print(f"Processing order {fields['order_id']}")                # Acknowledge and delete the entry (default behavior)        redis.xackdel("orders", "processors", entry_id)                # Or use DELREF for complete cleanup        # redis.xackdel("orders", "processors", entry_id, option="DELREF")

Batch Processing

# Read pending messagespending = redis.xreadgroup(    "tasks", "worker1", {"task-queue": ">"},    count=10)if pending:    stream_name, entries = pending[0]    processed_ids = []        # Process each task    for entry_id, fields in entries:        try:            # Process task...            print(f"Processing task {entry_id}")            processed_ids.append(entry_id)        except Exception as error:            print(f"Failed to process {entry_id}: {error}")        # Acknowledge and delete all successfully processed tasks    # Using DELREF to completely remove references    if processed_ids:        result = redis.xackdel("task-queue", "tasks", *processed_ids, option="DELREF")        print(f"Results: {result}")  # List of results for each ID

Audit Mode with ACKED

# Process messages but keep them for audit/debugging purposesmessages = redis.xreadgroup(    "audit", "auditor1", {"events": ">"},    count=5)for stream_name, entries in messages:    entry_ids = [entry_id for entry_id, _ in entries]        # Process and acknowledge but don't delete    for entry_id, fields in entries:        # Audit the event...        print(f"Auditing event {fields}")        # Acknowledge without deleting (keep for audit trail)    result = redis.xackdel("events", "audit", *entry_ids, option="ACKED")    print(f"Acknowledged {len(result)} events (kept in stream)")

Use Cases

Message Queue Cleanup: Process messages and remove them from the stream in one operation
Event Processing: Acknowledge event handling and clean up the stream
Task Queue Management: Complete tasks and remove them atomically
Memory Optimization: Reduce stream memory usage by removing processed entries

Comparison with XACK + XDEL

Traditional approach (two operations):

# Acknowledge the messageredis.xack("mystream", "mygroup", "123-0")# Then delete itredis.xdel("mystream", "123-0")

With XACKDEL (single atomic operation):

# Acknowledge and delete in one operation (default)redis.xackdel("mystream", "mygroup", "123-0")# With options for fine-grained controlredis.xackdel("mystream", "mygroup", "123-0", option="DELREF")  # Complete cleanupredis.xackdel("mystream", "mygroup", "123-0", option="ACKED")   # Acknowledge only

This command is available in Redis 8.2.0 and later. It combines XACK and XDEL into a single atomic operation, which is more efficient and ensures consistency.