# Celery Transport Concepts (/integrations/celery/concepts)
## The Kombu Virtual Transport [#the-kombu-virtual-transport]
Celery never talks to a broker directly. It delegates all messaging to [Kombu](https://docs.celeryq.dev/projects/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.
```python title="src/kubemq_celery/transport.py"
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:
```python title="src/kubemq_celery/transport.py"
implements = virtual.Transport.implements.extend(
asynchronous=False,
exchange_type=frozenset(["direct", "topic", "fanout"]),
heartbeats=False,
)
```
## Auto-Registration [#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.
```python title="src/kubemq_celery/__init__.py"
# 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"
```
`import kubemq_celery` must run **before** Celery resolves the broker URL. Without it, the `kubemq://` scheme is unknown and Celery raises an "unknown transport" error. The same `kubemq://` scheme is reused for the result backend — Celery distinguishes them by setting name, not by URL.
## Client IDs and Connections [#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.
```python title="src/kubemq_celery/transport.py"
"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**.
| Aspect | KubeMQ (gRPC) | Redis | RabbitMQ (AMQP) |
| ---------------------- | ------------------------------- | -------------------------------- | -------------------------- |
| Connections per worker | 2-3 (queues + pubsub + backend) | 6-8 (broker + backend + results) | 4-6 (channels + heartbeat) |
| Protocol | HTTP/2 multiplexed | TCP per connection | TCP per channel |
| Keepalive | Built-in gRPC keepalive | Manual PING/PONG | AMQP heartbeats |
## Acknowledgment Model [#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.
```python
# 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:
```python title="src/kubemq_celery/transport.py"
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)
```
With `task_acks_late=True`, tasks that run longer than \~60 seconds risk the KubeMQ server-side transaction timeout expiring before the ack is sent, which triggers redelivery. For long-running tasks, use `acks_early` (the default) or design tasks to be idempotent.
## Delayed Delivery [#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.
```python
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](/integrations/celery/how-to/scheduling) for longer schedules.
## Queue Name Sanitization [#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.
| Celery Name | KubeMQ Channel |
| ------------------------------ | ------------------------------ |
| `celery` | `celery` |
| `celery@worker1.celery.pidbox` | `celery.worker1.celery.pidbox` |
| `reply/celery/pidbox` | `reply.celery.pidbox` |
The rules: replace `@`, `/`, `#`, and the Redis priority separator (`\x06`) with `.`; replace spaces with `_`; collapse consecutive dots; and strip leading and trailing dots.
```python title="src/kubemq_celery/utils.py"
_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 [#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.
```python
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.
Result expiration caps at `86400` seconds (24 hours), a KubeMQ limitation. Celery's default `result_expires` of 24 hours already matches this maximum. For longer retention, use a database-backed result backend.
## Mapping Celery Primitives to KubeMQ [#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 [#known-limitations]
Keep the following constraints in mind when designing tasks for the KubeMQ transport:
* **Priority is metadata-only** — `task_default_priority` and `task_queue_max_priority` are stored in message tags but **not** enforced for ordering at the server level. Use separate queues with `task_routes` for true priority routing.
* **Max delay / expiration is 24 hours** — `delay_in_seconds`, `message_expiration`, and `result_expires` all cap at `86400` seconds. Larger values are capped with a warning log.
* **Chords use a polling fallback** — chord completion is tracked by Celery's `chord_unlock` task, which polls for group completion rather than relying on a native broker callback.
* **Long `acks_late` tasks risk redelivery** — tasks running longer than \~60 seconds with `task_acks_late=True` may exceed the KubeMQ transaction timeout and be redelivered. Prefer `acks_early` or idempotent tasks.
## Running a Local Broker [#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:
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](/integrations/celery/tutorials/getting-started) for the full walkthrough.
## Related Topics [#related-topics]