Celery Transport Concepts

The Kombu Virtual Transport

Celery never talks to a broker directly. It delegates all messaging to Kombu, Celery's transport abstraction layer. Kombu defines a virtual transport contract — a small set of storage primitives that any backing store can implement — and kubemq-celery implements that contract against KubeMQ.

The core class is Channel, which extends Kombu's virtual.Channel and implements the storage primitives _put() (send a task) and _get() (receive a task). Task messages flow over KubeMQ Queues; fanout traffic — pidbox remote control and Celery monitoring events — flows over KubeMQ Events, enabled by setting supports_fanout = True on the channel.

class Channel(BaseKubeMQChannel, virtual.Channel):
    """KubeMQ Channel -- implements storage primitives for Kombu virtual transport."""

    supports_fanout = True
    do_restore = False  # KubeMQ handles redelivery natively

    def _put(self, queue: str, message: dict, **kwargs: Any) -> None:
        """Send a Celery message to a KubeMQ Queue channel."""
        msg_kwargs = self._build_queue_message_kwargs(queue, message)
        msg = QueueMessage(**msg_kwargs)
        self._kubemq_queues_client.send_queue_message(msg)

    def _put_fanout(self, exchange: str, message: dict, routing_key: str = "", **kwargs):
        """Publish a message to a fanout exchange via KubeMQ Events."""
        body = json.dumps(message).encode("utf-8")
        event = EventMessage(channel=sanitize_queue_name(exchange), body=body)
        self._kubemq_pubsub_client.publish_event(event)

Direct and topic exchanges map onto KubeMQ Queues — routing is handled by the queue name. Fanout exchanges (pidbox, the celeryev monitoring exchange) map onto KubeMQ Events, where every subscribed worker or monitoring tool receives every message. The transport declares its supported exchange types explicitly:

implements = virtual.Transport.implements.extend(
    asynchronous=False,
    exchange_type=frozenset(["direct", "topic", "fanout"]),
    heartbeats=False,
)

Auto-Registration

Kombu and Celery resolve a broker URL scheme (kubemq://) and a result backend scheme to concrete Python classes through registries. Importing the package once registers both, which is why a single import kubemq_celery is all the wiring you need.

# Auto-register transport aliases
from kombu.transport import TRANSPORT_ALIASES

TRANSPORT_ALIASES["kubemq"] = "kubemq_celery.transport:Transport"
TRANSPORT_ALIASES["kubemq+tls"] = "kubemq_celery.transport:Transport"

# Auto-register result backend alias
from celery.app.backends import BACKEND_ALIASES

BACKEND_ALIASES["kubemq"] = "kubemq_celery.backend:KubeMQResultBackend"

Client IDs and Connections

Each Channel lazily creates up to two KubeMQ gRPC clients: a QueuesClient for task traffic and a PubSubClient for fanout. Both derive a unique client ID from the client_id_prefix transport option (default celery) plus a random suffix, so multiple workers never collide on the broker.

"client_id": f"{self.client_id_prefix}-{prefix}-{uuid4().hex[:8]}",

A worker therefore connects as celery-queues-{rand8} and celery-pubsub-{rand8}. Because gRPC runs over HTTP/2 and multiplexes many logical streams over a single TCP connection, a worker needs only 2-3 connections (queues, pubsub, and the result backend), compared to 6-8 for Redis or 4-6 for RabbitMQ.

Acknowledgment Model

KubeMQ provides native, message-level acknowledgment over the gRPC stream. There is no Redis-style visibility timeout — and therefore none of the duplicate-delivery races that come from setting that timeout too low or the reprocessing stalls from setting it too high. The transport maps Celery's two ack modes onto KubeMQ primitives:

• acks_early (default, task_acks_late=False) — the message is acknowledged on receipt. The transport passes auto_ack=True to the receive call, so the broker drops the message the moment it is delivered. This is the safest, most performant option, with zero ack overhead.
• acks_late (task_acks_late=True) — the message is acknowledged after the task completes successfully, using a single native gRPC ack() call. This gives at-least-once delivery: if the worker crashes mid-task, KubeMQ redelivers the message.

# acks_early (default) — auto-ack on receive, zero overhead
app.conf.task_acks_late = False

# acks_late — native ack after success, at-least-once delivery
app.conf.task_acks_late = True
app.conf.task_reject_on_worker_lost = True  # maps to KubeMQ nack()

Under the hood, basic_ack resolves the stored message reference and calls its native ack(); basic_reject calls re_queue() when requeue=True and nack() otherwise:

def basic_ack(self, delivery_tag: str, multiple: bool = False) -> None:
    msg_ref = self._kubemq_msg_refs.pop(delivery_tag)
    msg_ref.ack()
    super().basic_ack(delivery_tag, multiple)

Delayed Delivery

Celery's countdown and eta map onto KubeMQ's native delay_in_seconds field on a queue message. There is no client-side polling loop (as Redis requires) and no broker plugin (as RabbitMQ's rabbitmq_delayed_message_exchange requires) — the broker holds the message and releases it when the delay elapses.

task.apply_async(countdown=60)          # delivered after 60 seconds
task.apply_async(eta=future_datetime)   # delivered at a specific time

The maximum delay is 24 hours (86400 seconds). Values beyond that are capped at 24 hours with a warning log — use Celery Beat for longer schedules.

Queue Name Sanitization

KubeMQ channel names allow only [a-zA-Z0-9._-], which is stricter than Redis keys or AMQP queue names. Celery's internal names contain characters that are invalid in KubeMQ channels — most notably the @ in pidbox names and / in reply queues. The transport sanitizes every name transparently through sanitize_queue_name, so no configuration change is needed.

The rules: replace @, /, #, and the Redis priority separator (\x06) with .; replace spaces with _; collapse consecutive dots; and strip leading and trailing dots.

_SANITIZE_MAP = str.maketrans(
    {"@": ".", "/": ".", "#": ".", " ": "_", "\x06": "."}
)

def sanitize_queue_name(name: str) -> str:
    result = name.translate(_SANITIZE_MAP)
    result = re.sub(r"\.{2,}", ".", result)  # collapse dots
    result = result.strip(".")
    return result

Result Backend Mechanics

When the optional result backend is enabled (result_backend="kubemq://..."), task results are stored as KubeMQ Queue messages on a per-task channel named celery-result-{task_id}. Retrieval uses peek_queue_messages(), a non-destructive read — the message is not consumed, so multiple callers (or repeated result.get() calls) can read the same result.

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

As a task moves through its lifecycle, each state transition (PENDING → STARTED → SUCCESS) purges the result channel and rewrites the message with the new state, so a peek always returns the latest state. Because results live on the same broker as the task queues, no separate Redis or database is needed for a pure-KubeMQ stack.

Mapping Celery Primitives to KubeMQ

The diagram below shows how each Celery primitive is routed: task dispatch and results travel over KubeMQ Queues, while pidbox control and monitoring events use KubeMQ Events fanout.

Known Limitations

Running a Local Broker

The transport connects to KubeMQ over the native gRPC port 50000 — no HTTP connector flag is required (unlike connectors such as CloudEvents). Run a broker locally with Docker, exposing the gRPC port and the shared HTTP server:

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 Celery transport uses. Port 9090 is the shared HTTP server (REST and the /health probe) — curl http://localhost:9090/health confirms the broker is reachable. See Getting Started for the full walkthrough.

Related Topics