# Migrating from RabbitMQ (/connectors/rabbitmq/reference/migration-from-rabbitmq) RabbitMQ applications using any standard AMQP 0-9-1 client library (`pika`, `amqp091-go`, `php-amqplib`, Spring AMQP, `amqplib` for Node.js, …) can point at KubeMQ by changing **only the connection string** — no code changes — for the supported feature set. KubeMQ speaks the RabbitMQ wire dialect, so `queue.declare`, `exchange.declare`, publisher confirms, DLX, and Direct Reply-To work as-is. This is an **endpoint-only** drop-in: same library, same code, same protocol, with no KubeMQ SDK to adopt. ## Overview [#overview] | | | | -------------------- | ----------------------------------------------------------------------------------------------------------------------------------------- | | **Connector** | KubeMQ RabbitMQ connector (AMQP 0-9-1 RabbitMQ dialect) | | **Ports** | 5672 (AMQP plain) / 5671 (AMQPS/TLS) | | **Canonical client** | [`pika`](https://pika.readthedocs.io/) 1.x (Python) | | **Drop-in level** | Endpoint-only — change the connection string; keep all client library code unchanged for the supported feature set | | **Enable default** | Opt-in — `Connectors.Amqp.Enable = false` by default. Set `CONNECTORS_AMQP_ENABLE=true` (or `Enable = true` in TOML) to open the listener | The RabbitMQ connector is **disabled by default** — a stock kubemq-server does not bind the AMQP listener until you turn it on: **The enable variable is `CONNECTORS_AMQP_ENABLE`** — this is the **AMQP 0-9-1 (RabbitMQ)** connector, distinct from the AMQP 1.0 connector's `CONNECTORS_AMQP10_ENABLE`. For Kubernetes, set `spec.amqp.enabled: true` in the `KubemqCluster` CR. ## Compatibility Matrix [#compatibility-matrix] The cells below are the RabbitMQ column of the [migration hub](/connectors/how-to/migration)'s master cross-protocol matrix. | Dimension | RabbitMQ on KubeMQ | | -------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | **Point-to-point queues** | ✅ | | **Pub/sub (non-durable)** | ✅ exchanges | | **Durable / persistent subscriptions** | ✅ durable queues | | **Request / reply (RPC)** | ✅ Direct Reply-To | | **Ordering guarantee** | ✅ per-queue¹ | | **Transactions** | ❌ (use publisher confirms) | | **Dead-letter / redrive** | ✅ DLX + x-death² | | **Selectors / filtering / wildcards** | ✅ topic/headers routing | | **Auth model** | SASL PLAIN (JWT) | | **TLS / mTLS** | ✅ 5671 | | **Top unsupported** | See [Hard rejections](#hard-rejections-connectionclose-540-not-implemented) and [Accepted-but-inert arguments](#accepted-but-inert-arguments-stored-never-applied) below | ¹ Requeued messages re-enter at the queue **tail**, not near the head — see [Requeue → tail](#what-does-not-migrate--deviations). ² Dead-lettering fires on the `rejected` trigger only; the `expired` trigger does not exist. Poison messages exceeding `MaxReceiveCount` are silently dropped — there is no consumable dead-letter address for those over this protocol. ## Connection / Endpoint Migration [#connection--endpoint-migration] Replace the RabbitMQ host and port with the KubeMQ host. The URI scheme, client library, and application code stay the same: ```text title="AMQP URI swap" # Before (RabbitMQ) amqp://user:password@rabbitmq.example.com:5672/ amqps://user:password@rabbitmq.example.com:5671/myvhost # After (KubeMQ) amqp://user:password@kubemq.example.com:5672/ amqps://user:password@kubemq.example.com:5671/myvhost ``` Key differences to be aware of at connection time: | Aspect | RabbitMQ | KubeMQ | | --------------- | ---------------------------- | --------------------------------------------------------------------------------------------------------------------------------- | | SASL mechanism | PLAIN, AMQPLAIN, EXTERNAL, … | **PLAIN only** | | Password | RabbitMQ user password | **KubeMQ JWT** when `Authentication.Enable = true`; any value when auth is disabled (username recorded for audit) | | Vhost | Must be pre-created | **Namespace-on-first-use** — any charset-valid vhost is accepted; `/` maps to the configured `DefaultVhost` (default `"default"`) | | Client identity | Connection name (optional) | `amqp-{connection_name\|uuid8}` — used for Casbin authorization; policies must cover `amqp-*` IDs | | TLS | Per-listener config | Active when the server's `Security` block is configured; serves the same certificates as gRPC/REST | | Queue storage | RabbitMQ queues | KubeMQ Queue channels `amqp.{vhost}.{queue}` | ## Concept & Destination Mapping [#concept--destination-mapping] AMQP's exchange/binding model is implemented connector-side as virtual routing. Only queues hold messages; exchanges and bindings are metadata resolved at publish time. | RabbitMQ concept | KubeMQ pattern | Channel / address | | -------------------------------------------- | --------------------------- | -------------------------- | | Queue | Queues | `amqp.{vhost}.{queue}` | | Exchange (direct / fanout / topic / headers) | Virtual routing to queues | — | | Binding | Routing rule (vhost-scoped) | — | | Dead-letter exchange (DLX) | Dead-letter re-routing | Target queue channel | | Direct Reply-To (`amq.rabbitmq.reply-to`) | In-connector reply shortcut | Opaque per-channel address | | Vhost `/` | `DefaultVhost` segment | `amqp.default.{queue}` | | Vhost `myvhost` | Literal segment | `amqp.myvhost.{queue}` | A message published over AMQP to queue `orders` in the default vhost lands in the KubeMQ Queue channel `amqp.default.orders`. That channel is interoperable with gRPC/REST queue clients: an AMQP producer can feed a native KubeMQ consumer on the same channel, and vice-versa. See [Channel mapping](/connectors/rabbitmq/reference/channel-mapping) for the full grammar. ## Canonical Client Example [#canonical-client-example] The examples below use [`pika`](https://pika.readthedocs.io/) 1.x. Key API symbols: `pika.BlockingConnection`, `pika.URLParameters`, `pika.ConnectionParameters`, `channel.queue_declare`, `channel.basic_publish`, `channel.basic_consume`, `channel.basic_ack`, `channel.basic_get`. ### Work queue (publish + consume) [#work-queue-publish--consume] ```python import pika # pika 1.x params = pika.URLParameters("amqp://user:YOUR_JWT@kubemq.example.com:5672/") conn = pika.BlockingConnection(params) ch = conn.channel() ch.queue_declare(queue="orders", durable=True) # Publish ch.basic_publish( exchange="", routing_key="orders", body=b'{"order_id": "ORD-1"}', ) # Consume (manual ack) def on_message(ch, method, props, body): print("received:", body) ch.basic_ack(delivery_tag=method.delivery_tag) ch.basic_qos(prefetch_count=10) ch.basic_consume(queue="orders", on_message_callback=on_message) ch.start_consuming() ``` ### Topic exchange routing [#topic-exchange-routing] ```python ch.exchange_declare(exchange="logs", exchange_type="topic") ch.queue_declare(queue="errors", durable=True) ch.queue_bind(exchange="logs", queue="errors", routing_key="*.error") ch.basic_publish(exchange="logs", routing_key="app.error", body=b"boom") # routed ch.basic_publish(exchange="logs", routing_key="app.info", body=b"fyi") # not routed ``` ### Publisher confirms [#publisher-confirms] ```python ch.confirm_delivery() ch.basic_publish( exchange="", routing_key="orders", body=b"payload", mandatory=True, ) # pika raises UnroutableError on basic.return when mandatory=True and no route ``` ### Dead-letter exchange [#dead-letter-exchange] ```python ch.exchange_declare(exchange="dlx", exchange_type="fanout") ch.queue_declare(queue="dead", durable=True) ch.queue_bind(exchange="dlx", queue="dead", routing_key="") ch.queue_declare( queue="work", durable=True, arguments={"x-dead-letter-exchange": "dlx"}, ) # A consumer that basic_nack(..., requeue=False) on "work" dead-letters to "dead" # with RabbitMQ-exact x-death headers appended. ``` ### RPC (Direct Reply-To) [#rpc-direct-reply-to] ```python import uuid reply_queue = "amq.rabbitmq.reply-to" corr_id = str(uuid.uuid4()) # Responder (subscribe to the work queue and reply) def handle_request(ch, method, props, body): response = process(body) ch.basic_publish( exchange="", routing_key=props.reply_to, properties=pika.BasicProperties(correlation_id=props.correlation_id), body=response, ) ch.basic_ack(delivery_tag=method.delivery_tag) # Requester ch.basic_consume(queue=reply_queue, on_message_callback=on_reply, auto_ack=True) ch.basic_publish( exchange="", routing_key="rpc_queue", properties=pika.BasicProperties( reply_to=reply_queue, correlation_id=corr_id, ), body=b"hello", ) ``` ### One-shot pull (basic.get) [#one-shot-pull-basicget] ```python method, props, body = ch.basic_get(queue="orders", auto_ack=False) if method: print("got:", body) ch.basic_ack(delivery_tag=method.delivery_tag) else: print("queue empty") # note: ~1s latency floor on empty queue (deviation) ``` ## Security [#security] **Authentication — opt-in, default disabled.** `Connectors.Amqp.Enable` defaults to `false`. When you enable the connector, the authentication posture depends on whether `Authentication.Enable` is set: * **Auth disabled** (default): the listener accepts **any** SASL PLAIN credentials. If KubeMQ is reachable from untrusted networks, either enable authentication or firewall ports 5672/5671. * **Auth enabled**: the SASL PLAIN **password must be a valid KubeMQ JWT**. JWT validation happens at connect time only — token expiry does not terminate an established connection. The username is recorded for audit and `user-id` checks. **Authorization.** When Casbin is enabled, per-operation checks run against the mapped channel name (`amqp.{vhost}.{queue}`). Add policies for the `amqp-*` client IDs: ```text title="Casbin policies" # Allow all AMQP clients to publish to the orders queue in the default vhost allow amqp-.* Write amqp.default.orders # Allow consumers allow amqp-.* Read amqp.default.orders ``` Publish denials silently remove the denied target from the routed set (audited as `amqp.publish.denied`). Consume/topology denials return `channel.close 403`. **TLS.** Configure the server's `Security` block; the AMQPS listener on 5671 then serves the same certificates as gRPC/REST. Set `Connectors.Amqp.Port = 0` to force TLS-only AMQP. **Environment variables:** | Variable | Purpose | | -------------------------- | ---------------------------------------------------- | | `CONNECTORS_AMQP_ENABLE` | `true` to open the listener | | `CONNECTORS_AMQP_PORT` | Plain TCP listener port (default 5672; `0` disables) | | `CONNECTORS_AMQP_TLS_PORT` | TLS listener port (default 5671; `0` disables) | See [Authentication & security](/connectors/reference/auth-and-security) for JWT issuance and Casbin policy syntax. ## What Does NOT Migrate / Deviations [#what-does-not-migrate--deviations] The connector covers the common AMQP 0-9-1 feature set, but a handful of methods are rejected outright, a handful of declare arguments are accepted but never applied, and a handful of behaviors differ intentionally. The three lists below are **distinct** — the first list errors loudly, the second silently no-ops, and the third changes observable behavior. ### Hard rejections (`connection.close 540 not-implemented`) [#hard-rejections-connectionclose-540-not-implemented] These AMQP methods are rejected immediately — your application will receive a protocol-level error and must stop using them: | Feature | Behavior | | ------------------------------------------------------------------- | ---------------------------------------------------------- | | Transactions (`tx.select`, `tx.commit`, `tx.rollback`) | `540 not-implemented` — use **publisher confirms** instead | | Exchange-to-exchange bindings (`exchange.bind` / `exchange.unbind`) | `540`; capability advertised `false` | | `connection.update-secret` | `540` | | `basic.recover-async` | `540` | | `immediate=true` on `basic.publish` | `540` | ### Accepted-but-inert arguments (stored, never applied) [#accepted-but-inert-arguments-stored-never-applied] The following queue/exchange declare arguments are accepted by the server, stored in topology metadata, surfaced in the dashboard, and logged once per entity — they are **silently stored and never alter behavior**. If your application relies on these for routing or flow control, that logic must move into the application: * Alternate exchange (`x-alternate-exchange`) * Queue length limits and overflow (`x-max-length`, `x-max-length-bytes`, `x-overflow`) * Queue expiry (`x-expires`) * Queue type (`x-queue-type`), lazy mode (`x-queue-mode`), single-active-consumer (`x-single-active-consumer`) * Message priority (`x-max-priority`) These arguments **do not error** — they are accepted and ignored. A queue declared with `x-max-length` or `x-max-priority` behaves like an ordinary queue; the limit or priority is never enforced. Audit your declares for any reliance on these before you cut over. ### Behavioral deviations [#behavioral-deviations] These behaviors differ intentionally from RabbitMQ. Review each against your application before migrating. 1. **Requeue → tail.** Requeued messages re-enter at the queue **tail**. RabbitMQ classic queues preserve near-head position. 2. **TTL expiry is an eager drop.** Expired messages are silently dropped inside the broker — they are **never** dead-lettered. The RabbitMQ `expired` DLX trigger does not fire. 3. **TTL clamped.** Per-message TTL is clamped to `Queue.MaxExpirationSeconds` (default 12h); `x-delay` is clamped to `MaxDelaySeconds`. 4. **`basic.get` latency floor.** `basic.get` on an empty queue has a \~1s latency floor (KubeMQ minimum wait). RabbitMQ returns `get-empty` immediately. Prefer `basic.consume`. 5. **`MaxReceiveCount` drop.** Messages redelivered beyond the broker `MaxReceiveCount` (default 1024) are dropped or broker-rerouted. RabbitMQ redelivers forever. 6. **DLX trigger — rejected only.** DLX fires on the `rejected` trigger (`basic.reject` / `nack` with `requeue=false`) only. The `x-death` header is synthesized for broker-rerouted poison messages. The `expired` trigger does not exist (see deviation 2). 7. **Inert queue arguments.** Priority, max-length/overflow, `x-expires`, queue-type, and single-active-consumer arguments are inert (see [Accepted-but-inert arguments](#accepted-but-inert-arguments-stored-never-applied) above). 8. **Node-local exclusivity.** Exclusive queues and Direct Reply-To are **node-local** in cluster mode. Exclusive queues from two nodes share the channel name; Direct Reply-To requires the requester and responder on the same node (use load-balancer session affinity). 9. **Reserved vhost name.** The literal vhost name `default` (the configured `DefaultVhost` value) is reserved — reach it via `/`. Queue and vhost names must not contain `; : * >`, whitespace, or a trailing `.` (`406` / `402`). 10. **Partial-routing nack.** Publisher-confirm `basic.nack` on partial routing failure does **not** roll back queues that already accepted the message. A publisher retry may duplicate messages. See [Capabilities](/connectors/rabbitmq/reference/capabilities) and [Error codes](/connectors/rabbitmq/reference/error-codes) for the exhaustive per-method error tables and wire-contract detail. ## Verification Smoke Test [#verification-smoke-test] Use the work-queue snippet above as a publish-one / consume-one confirmation. 1. **Enable the connector.** Set `CONNECTORS_AMQP_ENABLE=true` and start (or restart) KubeMQ. Confirm the log line `started insecure amqp listener` (plain TCP) or `started secure amqp listener` (TLS) appears. 2. **Publish a message.** ```python import pika conn = pika.BlockingConnection(pika.URLParameters("amqp://user:pass@kubemq:5672/")) ch = conn.channel() ch.queue_declare(queue="smoke-test", durable=True) ch.basic_publish(exchange="", routing_key="smoke-test", body=b"hello-kubemq") conn.close() print("published ok") ``` 3. **Consume the message.** ```python import pika conn = pika.BlockingConnection(pika.URLParameters("amqp://user:pass@kubemq:5672/")) ch = conn.channel() method, props, body = ch.basic_get(queue="smoke-test", auto_ack=True) assert body == b"hello-kubemq", f"unexpected body: {body!r}" conn.close() print("consume ok:", body) ``` A successful publish followed by a matching consume confirms the connector is reachable and the queue channel is live. ## See Also [#see-also]