# Configuration Reference (/integrations/celery/reference/configuration)



This page is the configuration reference for the `kubemq-celery` package: broker URL schemes, the public API surface, the Celery settings it honors, environment variables, monitoring and control commands, known limitations, and the exceptions it raises. The full `broker_transport_options` and `result_backend_transport_options` key tables live on [Transport Options](/integrations/celery/reference/transport-options). For a guided introduction see [Getting Started](/integrations/celery/tutorials/getting-started); for worked configuration recipes see the [Configuration guide](/integrations/celery/how-to/configuration).

## Package facts [#package-facts]

| Field        | Value             |
| ------------ | ----------------- |
| Package name | `kubemq-celery`   |
| Version      | `1.1.0`           |
| License      | MIT               |
| Python       | `>= 3.10`         |
| Celery       | `>= 5.4`          |
| Kombu        | `>= 5.4`          |
| KubeMQ SDK   | `kubemq >= 4.1.5` |

Install from PyPI with `pip` or [uv](https://docs.astral.sh/uv/):

```bash
pip install kubemq-celery
# or
uv add kubemq-celery
```

The three runtime dependencies (`kubemq`, `celery`, `kombu`) are pulled in automatically. Optional extras for the example apps are listed under [Optional dependency extras](#optional-dependency-extras).

## Public API [#public-api]

The package exposes a deliberately small surface. Importing `kubemq_celery` does two things: it makes `Transport` and `KubeMQResultBackend` importable, and — as a side effect of the import — it registers the `kubemq` URL schemes with Kombu and the `kubemq` backend alias with Celery.

```python
import kubemq_celery

kubemq_celery.__version__          # "1.1.0"
kubemq_celery.Transport            # the Kombu transport class
kubemq_celery.KubeMQResultBackend  # the queue-peek result backend
```

| Name                  | Kind  | Description                                                                                            |
| --------------------- | ----- | ------------------------------------------------------------------------------------------------------ |
| `Transport`           | class | Kombu transport powering `kubemq://` broker URLs. Registered as the `kubemq` and `kubemq+tls` aliases. |
| `KubeMQResultBackend` | class | Queue-peek result backend. Registered as the `kubemq` backend alias.                                   |
| `__version__`         | str   | The installed package version (`"1.1.0"`).                                                             |

<Callout type="warn">
  The transport and backend aliases are registered *as a side effect of the import*. `import kubemq_celery` must run before Celery resolves the broker URL, or Celery raises `ValueError: Unknown transport 'kubemq'`. Keep the import at the top of your app module even if your editor flags it as unused.
</Callout>

Under the hood the import registers the following aliases:

```python title="src/kubemq_celery/__init__.py"
from kombu.transport import TRANSPORT_ALIASES

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

from celery.app.backends import BACKEND_ALIASES

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

If you cannot guarantee import order, set the result backend by its fully-qualified path instead of the alias: `result_backend = "kubemq_celery.backend:KubeMQResultBackend"`.

## Broker URL schemes [#broker-url-schemes]

The transport accepts four URL schemes. They differ in two dimensions: TLS on or off, and the synchronous gRPC client versus the native asyncio client.

```text
kubemq://[:token@]host[:port]
kubemq+tls://[:token@]host[:port]
kubemq+async://[:token@]host[:port]
kubemq+async+tls://[:token@]host[:port]
```

| Scheme                | TLS | Client                                           | Use with                                |
| --------------------- | --- | ------------------------------------------------ | --------------------------------------- |
| `kubemq://`           | No  | Synchronous                                      | Default worker pools (prefork, threads) |
| `kubemq+tls://`       | Yes | Synchronous                                      | Encrypted gRPC connections              |
| `kubemq+async://`     | No  | Async (`AsyncQueuesClient`, `AsyncPubSubClient`) | `--pool=asyncio` workers                |
| `kubemq+async+tls://` | Yes | Async                                            | Async pools over TLS                    |

The `+async` schemes use native async KubeMQ clients for non-blocking I/O and are intended for Celery's asyncio worker pool (`celery -A myapp worker --pool=asyncio`).

### URL components [#url-components]

| Component | Description                                                                   | Default     |
| --------- | ----------------------------------------------------------------------------- | ----------- |
| Scheme    | One of `kubemq://`, `kubemq+tls://`, `kubemq+async://`, `kubemq+async+tls://` | Required    |
| Token     | Authentication token, placed after `:` and before `@`                         | None        |
| Host      | KubeMQ broker hostname                                                        | `localhost` |
| Port      | KubeMQ gRPC port                                                              | `50000`     |

```python
# Basic connection (localhost, default gRPC port 50000)
app.conf.broker_url = "kubemq://localhost:50000"

# In-cluster Kubernetes service
app.conf.broker_url = "kubemq://kubemq.default.svc:50000"

# With an authentication token
app.conf.broker_url = "kubemq://:my-secret-token@kubemq.default.svc:50000"

# TLS, with and without a token
app.conf.broker_url = "kubemq+tls://kubemq.default.svc:50000"
app.conf.broker_url = "kubemq+tls://:my-token@kubemq.default.svc:50000"

# Async transport, with and without TLS
app.conf.broker_url = "kubemq+async://localhost:50000"
app.conf.broker_url = "kubemq+async+tls://kubemq.default.svc:50000"
```

A broker is reachable on the gRPC port `50000`. The quickest way to get one locally is Docker — the shared HTTP server (REST/health) is exposed on `9090`:

<RunKubeMQ ports="[50000, 9090]" />

## Celery settings compatibility [#celery-settings-compatibility]

The following table records how `kubemq-celery` treats common Celery settings. **Full** means the setting behaves as it does on any other broker; **Ignored** means the transport disregards it; **Stored** means the value is preserved on the message but not enforced server-side.

| Celery setting                    | Support | Notes                                                                              |
| --------------------------------- | ------- | ---------------------------------------------------------------------------------- |
| `broker_url`                      | Full    | `kubemq://`, `kubemq+tls://`, `kubemq+async://`, `kubemq+async+tls://` schemes.    |
| `broker_transport_options`        | Full    | All keys in [Transport Options](/integrations/celery/reference/transport-options). |
| `broker_pool_limit`               | Ignored | One client per Channel; Kombu's connection pool handles scaling.                   |
| `broker_connection_timeout`       | Full    | Passed to the SDK connection timeout.                                              |
| `broker_connection_retry`         | Full    | Standard Celery reconnection behavior.                                             |
| `broker_connection_max_retries`   | Full    | Standard Celery reconnection behavior.                                             |
| `broker_failover_strategy`        | Ignored | Not supported in this release; the KubeMQ cluster handles HA internally.           |
| `task_acks_late`                  | Full    | See the caveat below for long-running tasks.                                       |
| `task_acks_on_failure_or_timeout` | Full    | Standard Celery behavior.                                                          |
| `task_reject_on_worker_lost`      | Full    | Maps to KubeMQ `nack()`.                                                           |
| `task_default_queue`              | Full    | Queue name is sanitized for KubeMQ compatibility.                                  |
| `task_routes`                     | Full    | The virtual exchange layer handles routing.                                        |
| `task_default_priority`           | Stored  | Priority is stored in message tags, not enforced server-side.                      |
| `task_queue_max_priority`         | Stored  | Priority is stored in message tags.                                                |
| `worker_prefetch_multiplier`      | Full    | Managed by Kombu's virtual QoS layer.                                              |
| `worker_concurrency`              | Full    | Standard Celery behavior.                                                          |
| `result_backend`                  | Full    | `kubemq://` for the queue-peek result backend.                                     |
| `result_expires`                  | Full    | Controls result message expiration (max 24 hours).                                 |

<Callout type="info">
  **`task_acks_late` caveat.** With `task_acks_late=True`, a message is acknowledged only after the task completes. For tasks shorter than 60 seconds this works as expected. For tasks longer than 60 seconds, the KubeMQ server-side transaction timeout may expire before the task finishes, causing redelivery. For long-running tasks, either keep `task_acks_late=False` (the default, which uses `auto_ack=True` on receive) or accept at-least-once semantics and make the task idempotent.
</Callout>

## Environment variables [#environment-variables]

The example applications read connection settings from the environment. These are conventions used by the example code — your own app can wire them up however you prefer.

| Variable                | Default                    | Description                                                    |
| ----------------------- | -------------------------- | -------------------------------------------------------------- |
| `CELERY_BROKER_URL`     | `kubemq://localhost:50000` | KubeMQ broker URL.                                             |
| `CELERY_RESULT_BACKEND` | `kubemq://localhost:50000` | Result backend URL.                                            |
| `CELERY_ENV`            | `development`              | Environment name; affects logging and config in some examples. |

```python
import os

import kubemq_celery  # noqa: F401
from celery import Celery

app = Celery(
    "myapp",
    broker=os.environ.get("CELERY_BROKER_URL", "kubemq://localhost:50000"),
    result_backend=os.environ.get("CELERY_RESULT_BACKEND", "kubemq://localhost:50000"),
)
```

## Monitoring and control [#monitoring-and-control]

All of Celery's monitoring and control commands work over KubeMQ Events (pidbox fanout), so the standard tooling needs no changes beyond the broker URL.

### celery inspect [#celery-inspect]

```bash
# Check whether workers are alive (uses the transport's verify_connection ping)
celery -A myapp inspect ping

# List running tasks
celery -A myapp inspect active

# Worker pool statistics (uptime, task counts, prefetch)
celery -A myapp inspect stats

# Queues each worker is currently consuming
celery -A myapp inspect active_queues
```

### celery control [#celery-control]

Runtime control commands are broadcast to all online workers via Events fanout. They are fire-and-forget and are not persisted, so workers must be online to receive them.

```bash
# Add or remove a consumer at runtime
celery -A myapp control add_consumer new-queue
celery -A myapp control cancel_consumer old-queue

# Grow or shrink the worker pool
celery -A myapp control pool_grow 2
celery -A myapp control pool_shrink 1

# Set a runtime rate limit
celery -A myapp control rate_limit myapp.tasks.add 10/m
```

The same commands are available programmatically through `app.control`:

```python
app.control.ping(timeout=5)
app.control.inspect().active()
app.control.add_consumer("new-queue")
app.control.shutdown()
```

### Flower [#flower]

Flower works against a `kubemq://` broker like any other:

```bash
celery -A myapp flower --broker=kubemq://localhost:50000
```

### KubeMQ dashboard [#kubemq-dashboard]

The KubeMQ Management API dashboard on port `8080` shows broker-level metrics — queue depth, send/receive rates, and per-channel statistics. (The shared HTTP server on `9090` serves REST and the `/health` probe.) Inside a cluster, port-forward to reach the dashboard:

```bash
kubectl port-forward svc/kubemq 8080:8080
# then open http://localhost:8080
```

## Known limitations [#known-limitations]

| Limitation                  | Detail                                                                                                                                                                       |
| --------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Priority is metadata-only   | KubeMQ does not enforce message priority at the server level; `task_default_priority` and `task_queue_max_priority` are stored in message tags but not honored for ordering. |
| Maximum delay               | `countdown` / `eta` delays are capped at `86400` seconds (24 hours); larger values are clamped with a warning log.                                                           |
| Maximum result expiry       | `result_expires` and `message_expiration` are capped at `86400` seconds (24 hours), a KubeMQ limitation.                                                                     |
| Chord polling fallback      | Chords use Celery's `chord_unlock` task to poll for group completion rather than a native server-side primitive.                                                             |
| Long `task_acks_late` tasks | Tasks running longer than the KubeMQ transaction timeout (\~60s) under `task_acks_late=True` may be redelivered; design for idempotency or keep acks early.                  |
| `broker_failover_strategy`  | Not supported; the KubeMQ cluster handles HA internally.                                                                                                                     |

## Exceptions [#exceptions]

The transport defines an exception hierarchy in `kubemq_celery.exceptions`, all rooted at the KubeMQ SDK's `KubeMQError`. The base for this package is `KubeMQCeleryError`, with transport, backend, serialization, and config subclasses.

```python
from kubemq_celery.exceptions import (
    KubeMQCeleryError,            # base for all kubemq-celery errors
    KubeMQCeleryTransportError,   # send / receive / subscribe failures
    KubeMQCeleryConnectionError,  # broker connection failed or lost
    KubeMQCeleryChannelError,     # queue not found, permission denied
    KubeMQCeleryTimeoutError,     # operation timed out
    KubeMQCeleryBackendError,     # result backend store / retrieve errors
    KubeMQCelerySerializationError,  # JSON (de)serialization errors
    KubeMQCeleryConfigError,      # invalid transport configuration
)
```

You will also see these surfaced from the underlying KubeMQ SDK during connection problems:

| Exception                                           | When it appears                                                                                                                                                                                                                            |
| --------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| `KubeMQCeleryConnectionError`                       | The worker cannot reach the broker on startup (for example, "Connection refused").                                                                                                                                                         |
| `KubeMQAuthenticationError`                         | The auth token is missing or does not match the broker's configuration.                                                                                                                                                                    |
| `KubeMQStreamBrokenError` / `KubeMQConnectionError` | Intermittent loss of a live connection — network instability, a broker restart, or a load-balancer timeout. Configure gRPC keepalive (`grpc_keepalive_time`, `grpc_keepalive_timeout`, `grpc_permit_without_calls`) to detect and recover. |

For diagnosis steps see the [Troubleshooting](/integrations/celery/how-to/troubleshooting) guide.

## Optional dependency extras [#optional-dependency-extras]

The example apps in the repository require packages beyond the three runtime dependencies. They are declared under the `examples` extra in `pyproject.toml` and can be installed individually as needed.

```bash
# Framework integrations
pip install fastapi uvicorn   # FastAPI examples
pip install flask             # Flask examples
pip install starlette         # Starlette examples
pip install litestar          # Litestar examples
pip install aiohttp           # aiohttp examples
pip install django            # Django examples

# Serialization
pip install msgpack           # MessagePack serializer
pip install orjson            # Custom orjson serializer
```

| Package                 | Minimum version | Used by                                        |
| ----------------------- | --------------- | ---------------------------------------------- |
| `fastapi`               | `>= 0.100`      | FastAPI integration example                    |
| `uvicorn`               | `>= 0.20`       | ASGI server for FastAPI / Starlette / Litestar |
| `flask`                 | `>= 3.0`        | Flask integration example                      |
| `starlette`             | `>= 0.30`       | Starlette integration example                  |
| `litestar`              | `>= 2.0`        | Litestar integration example                   |
| `aiohttp`               | `>= 3.9`        | aiohttp integration example                    |
| `django`                | `>= 4.2`        | Django integration example                     |
| `django-celery-beat`    | `>= 2.5`        | Django periodic-task scheduling                |
| `django-celery-results` | `>= 2.5`        | Django result storage                          |
| `msgpack`               | `>= 1.0`        | MessagePack serializer example                 |

<Callout type="info">
  `orjson` is referenced by the custom-serializer example but is installed separately (`pip install orjson`); it is not part of the declared `examples` extra. Install only the packages an example actually needs rather than the whole extra.
</Callout>

## See also [#see-also]

<Cards>
  <Card title="Transport Options" href="/integrations/celery/reference/transport-options" description="Every broker_transport_options and result_backend_transport_options key, with type and default." />

  <Card title="Configuration guide" href="/integrations/celery/how-to/configuration" description="Worked examples for TLS/mTLS, DLQ, async transport, and gRPC keepalive." />

  <Card title="Getting Started" href="/integrations/celery/tutorials/getting-started" description="Run a Celery app on KubeMQ in about five minutes." />

  <Card title="Troubleshooting" href="/integrations/celery/how-to/troubleshooting" description="Diagnose connection, task, and result issues." />
</Cards>