Migrating from Redis or RabbitMQ

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

The entire change is one import and one broker URL:

# 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

# pip
pip install kubemq-celery

# uv (recommended)
uv add kubemq-celery

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

kubectl apply -f https://get.kubemq.io/deploy

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"

# If no longer needed
pip uninstall redis
# or
pip uninstall amqp

celery -A myapp worker --loglevel=info

Feature Comparison

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

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

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.

# 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

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.

# 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

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

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

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.

# 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

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.

Common Gotchas

New in v1.1

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):

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

Batch Receive

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

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

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://:

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

gRPC Keepalive

Tune keepalive for long-lived connections:

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

Redis-Specific

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

# 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

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

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

Next Steps