Result Backend

By default a Celery worker on KubeMQ runs with results disabled — tasks execute, but their return values are discarded. To capture and read those return values, enable the KubeMQ queue-peek result backend. It stores each result as a message on KubeMQ Queues and reads it back with a non-destructive peek, so you get a pure-KubeMQ stack with no external Redis instance or database to operate.

Enable the Result Backend

Add result_backend="kubemq://localhost:50000" alongside your broker URL. Both the broker and the backend can point at the same KubeMQ instance, so one broker serves the entire stack.

import kubemq_celery  # registers the kubemq:// transport
from celery import Celery

app = Celery(
    "myapp",
    broker="kubemq://localhost:50000",
    result_backend="kubemq://localhost:50000",
)

@app.task
def add(x, y):
    return x + y

You need a reachable KubeMQ broker. For local development, run one in Docker:

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

Port 50000 is the gRPC port the transport and result backend connect over. Port 9090 is the shared HTTP server (REST and the /health probe) — curl http://localhost:9090/health confirms the broker is up. The result backend talks to KubeMQ over native gRPC — no HTTP connector needs to be enabled.

How It Works

Each task result is written to its own KubeMQ Queue channel named celery-result-{task_id}. Retrieval uses peek_queue_messages(), a non-destructive read — the message stays on the channel after it is read, so multiple callers can fetch the same result independently without one consumer stealing it from another.

This is what lets a web request handler and a background poller both read the outcome of the same task, and it is also how Celery's group and chord machinery polls for completion.

Store and Retrieve a Result

With the backend enabled, dispatch a task with .delay() and block for its return value with result.get():

from tasks import add

result = add.delay(4, 6)
print(result.id)              # the task_id, also the result channel suffix
print(result.get(timeout=10))  # 10
print(result.state)           # SUCCESS

Because retrieval is a peek, you can re-read the same result later from anywhere that knows the task ID — construct an AsyncResult and call .get() again:

from celery.result import AsyncResult
from tasks import app

second_reader = AsyncResult(task_id, app=app)
print(second_reader.get(timeout=10))  # same value, message not consumed

Result Expiration

result_expires controls how long a stored result lives before KubeMQ removes it. KubeMQ caps message expiration at 86400 seconds (24 hours) — this is a KubeMQ limitation, and any larger value is silently capped. Celery's own default result_expires is also 24 hours, so the two line up with no surprises.

app.conf.update(
    result_backend="kubemq://localhost:50000",
    result_expires=86400,  # 24 hours — the KubeMQ maximum
)

result_expires accepts either an integer number of seconds or a timedelta:

from datetime import timedelta

app.conf.result_expires = timedelta(hours=1)  # 3600 seconds

The table below summarizes how each value maps to an effective expiration:

Result Backend Transport Options

The result backend opens its own KubeMQ client, configured independently of the broker via result_backend_transport_options. Use it to set authentication and TLS for the backend connection.

app.conf.update(
    result_backend="kubemq://localhost:50000",
    result_expires=3600,
    result_backend_transport_options={
        "auth_token": "my-token",
        "tls_enabled": False,
        "tls_cert_file": "/path/to/cert.pem",
        "tls_key_file": "/path/to/key.pem",
        "tls_ca_file": "/path/to/ca.pem",
    },
)

Task States

The result backend tracks a task across its lifecycle: PENDING → STARTED → SUCCESS. Each transition purges and rewrites the message on the celery-result-{task_id} channel, so a peek always returns the latest state rather than a backlog of historical entries.

To observe the STARTED state, enable task_track_started. Bound tasks can also publish a custom PROGRESS state with self.update_state(), which is written to the same channel:

app.conf.update(
    result_backend="kubemq://localhost:50000",
    task_track_started=True,  # required for the STARTED state
)

@app.task(bind=True)
def process_items(self, items: list[str]) -> dict:
    total = len(items)
    for i, item in enumerate(items):
        self.update_state(
            state="PROGRESS",
            meta={"current": i + 1, "total": total},
        )
        # ... process item ...
    return {"processed": total, "status": "completed"}

Clients poll the live state from the same AsyncResult:

result = process_items.delay(["alpha", "beta", "gamma"])
while not result.ready():
    print(result.state, result.info)  # PROGRESS, {"current": .., "total": ..}
print(result.get(timeout=10))         # final SUCCESS result

Fire-and-Forget Tasks

For tasks whose return value nobody reads — notifications, logging, cleanup — set ignore_result=True. The worker skips the backend entirely, so no celery-result-{task_id} channel is created and you avoid the overhead of writing a result that would only be discarded.

@app.task(ignore_result=True)
def send_notification(user_id: str, message: str) -> None:
    # No celery-result-{task_id} channel is created.
    # The caller cannot retrieve a return value.
    print(f"Notification sent to {user_id}: {message}")

A task marked ignore_result=True always reports result = None and stays in the PENDING state from the caller's perspective — never call .get() on it, as there is no result message to read.

Group Results and Custom Serializers

Group results work the same way. Dispatch a group of tasks in parallel and collect every return value with a single .get(); group metadata is stored on celery-group-{group_id} channels.

from celery import chord, group
from tasks import square, aggregate

# Group: run square(1..5) in parallel and collect all results
g = group(square.s(i) for i in range(1, 6))
print(g.apply_async().get(timeout=60))  # [1, 4, 9, 16, 25]

# Chord: run the group, then call aggregate(results) once all finish
c = chord([square.s(i) for i in range(1, 5)], aggregate.s())
print(c.apply_async().get(timeout=60))

Results are serialized with Celery's result_serializer. The default json is human-readable and cross-language; switch to pickle for full Python-type support or msgpack for a compact binary encoding. Workers and clients must agree on accept_content.

app.conf.update(
    result_serializer="json",   # or "pickle", "msgpack"
    task_serializer="json",
    accept_content=["json"],
)

Chord Dependency

KubeMQ does not provide native chord support, so kubemq-celery falls back to Celery's polling mechanism: the chord_unlock task repeatedly peeks the group result channels until every member completes, then fires the callback. This fallback requires the result backend to be configured — a chord with no result backend has nothing to poll and will never fire its callback.

Related