# Migrating from Redis or RabbitMQ (/integrations/celery/how-to/migration)



If you already run Celery on Redis or RabbitMQ, switching to KubeMQ is a configuration change, not a rewrite. `kubemq-celery` is a Kombu transport that registers the `kubemq://` URL scheme, so your task definitions, workers, canvas workflows, Beat schedules, and monitoring tooling keep working unchanged. This guide covers the one-line switch, a step-by-step rollout, what stays the same, what behaves differently, the gotchas worth knowing up front, and which Redis/RabbitMQ-specific settings to delete.

## One-Line Migration [#one-line-migration]

The entire change is one import and one broker URL:

```python title="tasks.py"
# Before (Redis):
app = Celery("myapp", broker="redis://localhost:6379/0")

# Before (RabbitMQ):
app = Celery("myapp", broker="amqp://guest:guest@localhost:5672//")

# After (KubeMQ):
import kubemq_celery  # registers the kubemq:// transport
app = Celery("myapp", broker="kubemq://localhost:50000")
```

That is it — one import and one URL change. The `import kubemq_celery` line registers the `kubemq://` scheme with Kombu before Celery resolves the broker URL.

## Step-by-Step Migration [#step-by-step-migration]

<Steps>
  <Step>
    ### Install kubemq-celery [#install-kubemq-celery]

    ```bash
    # pip
    pip install kubemq-celery

    # uv (recommended)
    uv add kubemq-celery
    ```
  </Step>

  <Step>
    ### Start a KubeMQ broker [#start-a-kubemq-broker]

    `kubemq-celery` talks to KubeMQ over native gRPC on port `50000` — no HTTP connector flag is required. Port `9090` is the shared HTTP server that hosts KubeMQ's REST, MCP, and other connectors; expose it if you want those endpoints alongside the broker.

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

    ```bash title="Kubernetes"
    kubectl apply -f https://get.kubemq.io/deploy
    ```
  </Step>

  <Step>
    ### Update your Celery configuration [#update-your-celery-configuration]

    ```python title="celeryconfig.py"
    import kubemq_celery  # add this import

    # Change broker_url
    app.conf.broker_url = "kubemq://localhost:50000"

    # Optional: move the result backend to KubeMQ too
    app.conf.result_backend = "kubemq://localhost:50000"
    ```
  </Step>

  <Step>
    ### Remove old broker dependencies (optional) [#remove-old-broker-dependencies-optional]

    Once nothing else in your app needs them, drop the old client libraries:

    ```bash
    # If no longer needed
    pip uninstall redis
    # or
    pip uninstall amqp
    ```
  </Step>

  <Step>
    ### Restart your workers [#restart-your-workers]

    ```bash
    celery -A myapp worker --loglevel=info
    ```

    Workers connect to KubeMQ and start processing tasks immediately.
  </Step>
</Steps>

## Feature Comparison [#feature-comparison]

| Feature                    | Redis                                     | RabbitMQ                            | KubeMQ                                              |
| -------------------------- | ----------------------------------------- | ----------------------------------- | --------------------------------------------------- |
| **Setup**                  | External process                          | External process + Erlang           | Docker or K8s-native                                |
| **Kubernetes-native**      | No                                        | No                                  | Yes (StatefulSet, auto-clustering)                  |
| **Message acknowledgment** | Visibility timeout (can cause duplicates) | Native ack/nack                     | Native ack/nack                                     |
| **Delayed delivery**       | Client-side polling                       | Plugin (`rabbitmq_delayed_message`) | Native `delay_in_seconds`                           |
| **Dead letter queue**      | Manual implementation                     | Exchange-based DLQ                  | Native `max_receive_count` + DLQ channel            |
| **Priority queues**        | Separate lists per priority               | Server-enforced priority            | Metadata tags (use separate queues for enforcement) |
| **Fanout / broadcast**     | Pub/Sub channels                          | Fanout exchange                     | KubeMQ Events                                       |
| **Monitoring (Flower)**    | Full                                      | Full                                | Full                                                |
| **Remote control**         | Full (pidbox via Redis Pub/Sub)           | Full (pidbox via AMQP)              | Full (pidbox via KubeMQ Events)                     |
| **Result backend**         | Redis `GET`/`SET`                         | RPC or DB                           | Queue-peek (non-destructive read)                   |
| **Protocol**               | TCP                                       | AMQP (TCP)                          | gRPC (HTTP/2)                                       |
| **Connection stability**   | Connection resets under load              | Stable                              | gRPC keep-alive, auto-reconnect                     |
| **Max delay**              | Unlimited (client polling)                | Unlimited (plugin)                  | 24 hours (86400 seconds)                            |
| **Max result expiry**      | Unlimited                                 | N/A (RPC: 24h default)              | 24 hours (86400 seconds)                            |

## What Works the Same [#what-works-the-same]

These Celery features behave identically with KubeMQ — no code changes:

* **Task definition** — `@app.task` decorators and task classes.
* **Task invocation** — `task.delay()`, `task.apply_async()`, and `task.s()` signatures.
* **Task routing** — `task_routes`, `task_default_queue`, and custom routing.
* **Canvas** — chains, groups, and chords; all canvas primitives work.
* **Task retries** — `self.retry()` and `autoretry_for`.
* **Worker management** — `celery worker`, `celery multi`, and concurrency settings.
* **Flower monitoring** — the full Flower feature set (task list, graphs, worker info).
* **Remote control** — `celery inspect` and `celery control` commands.
* **Celery Beat** — periodic task scheduling.
* **Serialization** — JSON, pickle, msgpack, and YAML serializers.
* **Prefetch** — `worker_prefetch_multiplier`, managed by Kombu's virtual QoS layer.

## What's Different [#whats-different]

The public Celery API is unchanged, but the mechanism behind several features differs. None of these require code changes; they explain behavior you may otherwise find surprising.

### Acknowledgment Model [#acknowledgment-model]

**Redis** uses a visibility timeout: if a worker crashes without acking within the timeout, the message becomes visible again. Set it too low and you get duplicate delivery; too high and reprocessing stalls. **KubeMQ** uses native ack/nack over a gRPC stream — messages are explicitly acknowledged or rejected, with no timing-based guesswork.

```python
# KubeMQ handles this natively — no configuration needed.
# These settings still work, but the underlying mechanism is different:
app.conf.task_acks_late = True               # KubeMQ native ack (not a visibility timeout)
app.conf.task_reject_on_worker_lost = True   # maps to KubeMQ nack()
```

### Delayed Tasks [#delayed-tasks]

**Redis** implements `countdown`/`eta` client-side with a polling loop. **RabbitMQ** requires the `rabbitmq_delayed_message_exchange` plugin. **KubeMQ** uses native `delay_in_seconds` on queue messages, with zero polling overhead.

```python
# Same API — the delivery mechanism is native:
task.apply_async(countdown=60)          # delivers after 60 seconds
task.apply_async(eta=future_datetime)   # delivers at a specific time

# Limitation: max delay is 24 hours (86400 seconds).
# Delays > 24h are capped at 24h with a warning log.
```

### Dead Letter Queue [#dead-letter-queue]

**Redis** requires a manual DLQ implementation; **RabbitMQ** requires exchange-level DLQ configuration. **KubeMQ** is a one-line transport option:

```python
app.conf.broker_transport_options = {
    "dead_letter_queue": "celery-dead-letters",
    "max_receive_count": 3,
}
```

### Result Backend [#result-backend]

**Redis** uses `GET`/`SET` with TTL expiry over a separate connection. **KubeMQ** uses a non-destructive queue peek: results are stored as Queue messages in the same broker, so multiple callers can read the same result and you need no additional infrastructure.

```python
# Pure KubeMQ stack — broker and results in one place
app.conf.result_backend = "kubemq://localhost:50000"
app.conf.result_expires = 86400  # max 24 hours
```

### Queue Name Sanitization [#queue-name-sanitization]

KubeMQ channel names have stricter character rules than Redis keys or AMQP queue names, so the transport sanitizes names automatically. This is transparent — no configuration changes needed.

| Celery Name                    | KubeMQ Channel                 |
| ------------------------------ | ------------------------------ |
| `celery`                       | `celery`                       |
| `celery@worker1.celery.pidbox` | `celery.worker1.celery.pidbox` |
| `reply/celery/pidbox`          | `reply.celery.pidbox`          |

## Common Gotchas [#common-gotchas]

<Accordions>
  <Accordion title="import kubemq_celery is required">
    The transport must be imported to register the `kubemq://` URL scheme. Without it, Celery raises an "unknown transport" error. Import it before Celery resolves the broker URL:

    ```python
    import kubemq_celery  # must run before Celery uses the broker URL
    from celery import Celery
    ```
  </Accordion>

  <Accordion title="broker_pool_limit is ignored">
    KubeMQ uses a single gRPC client per Channel, and Kombu's connection pool handles scaling. Setting `broker_pool_limit` has no effect.
  </Accordion>

  <Accordion title="Maximum delay is 24 hours">
    KubeMQ's `delay_in_seconds` caps at 86400 seconds. `countdown` or `eta` values beyond 24 hours are capped (with a warning log). Use Celery Beat for longer scheduling needs.
  </Accordion>

  <Accordion title="Result expiration is 24 hours max">
    KubeMQ queue message expiration caps at 86400 seconds. Celery's default `result_expires` of 24 hours matches the KubeMQ maximum. If you need longer result retention, use a database-backed result backend.
  </Accordion>

  <Accordion title="Priority is metadata only">
    KubeMQ does not enforce message priority at the server level — priority values are stored in message tags but not used for ordering. Use separate queues for priority routing instead:

    ```python
    app.conf.task_routes = {
        "myapp.tasks.critical_*": {"queue": "high-priority"},
        "myapp.tasks.batch_*": {"queue": "low-priority"},
    }
    ```
  </Accordion>

  <Accordion title="Long-running tasks with task_acks_late=True">
    With `task_acks_late=True`, tasks running longer than \~60 seconds may outlast the KubeMQ server-side transaction, which expires before the ack is sent and causes redelivery. For long tasks, use `task_acks_late=False` (the default) or make your tasks idempotent.
  </Accordion>
</Accordions>

## New in v1.1 [#new-in-v11]

### Per-Message TTL [#per-message-ttl]

Set a default message expiration for all tasks. A task-level `expires` header takes precedence, and the maximum is 86400 seconds (24 hours):

```python
app.conf.broker_transport_options = {
    "message_expiration": 3600,  # 1 hour TTL for all messages
}
```

### Batch Receive [#batch-receive]

Receive multiple messages per gRPC call to cut round-trips and raise throughput. `max_batch_size` accepts up to 100:

```python
app.conf.broker_transport_options = {
    "max_batch_size": 10,  # up to 100
}
```

### Async Transport [#async-transport]

For asyncio-based worker pools (Starlette, Litestar, and similar), use the async transport scheme and the asyncio pool. TLS is also supported via `kubemq+async+tls://`:

```python
app.conf.broker_url = "kubemq+async://localhost:50000"
# Start the worker with: celery -A myapp worker --pool=asyncio
```

### gRPC Keepalive [#grpc-keepalive]

Tune keepalive for long-lived connections:

```python
app.conf.broker_transport_options = {
    "grpc_keepalive_time": 30,        # ping every 30s
    "grpc_keepalive_timeout": 10,     # wait 10s for a response
    "grpc_permit_without_calls": True,
}
```

## Settings to Remove [#settings-to-remove]

### Redis-Specific [#redis-specific]

These Redis-specific settings have no equivalent on KubeMQ and can be deleted:

```python
# Remove these:
# app.conf.redis_max_connections = ...
# app.conf.redis_socket_timeout = ...
# app.conf.redis_socket_connect_timeout = ...
# app.conf.redis_retry_on_timeout = ...
# app.conf.broker_transport_options = {"visibility_timeout": 3600}  # not needed with KubeMQ
```

### RabbitMQ-Specific [#rabbitmq-specific]

These RabbitMQ-specific settings are not applicable — exchange and queue declaration is handled automatically:

```python
# Remove these:
# app.conf.broker_heartbeat = ...
# app.conf.broker_transport_options = {"confirm_publish": True}  # not applicable
```

<Callout type="info">
  Most other Celery settings carry over directly. For the exact support level of each key — including which options are `Full`, `Stored` (priority tags), or `Ignored` (such as `broker_pool_limit` and `broker_failover_strategy`) — see the [Reference](/integrations/celery/reference/configuration).
</Callout>

## Next Steps [#next-steps]

<Cards>
  <Card title="Configuration" href="/integrations/celery/how-to/configuration" description="All broker transport options, TLS/mTLS, async, and result backend settings." />

  <Card title="Kubernetes Deployment" href="/integrations/celery/how-to/kubernetes" description="Run the broker and Celery workers in-cluster with auto-clustering and KEDA autoscaling." />

  <Card title="Reference" href="/integrations/celery/reference/configuration" description="URL schemes, transport options, and Celery settings compatibility." />
</Cards>