Migrating 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

The RabbitMQ connector is disabled by default — a stock kubemq-server does not bind the AMQP listener until you turn it on:

docker run -d \  --name kubemq \  -p 5672:5672 \  -p 5671:5671 \  -p 50000:50000 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  -e CONNECTORS_AMQP_ENABLE=true \  europe-docker.pkg.dev/kubemq/images/kubemq:next

Compatibility Matrix

The cells below are the RabbitMQ column of the migration hub's master cross-protocol matrix.

¹ Requeued messages re-enter at the queue tail, not near the head — see Requeue → tail.

² 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

Replace the RabbitMQ host and port with the KubeMQ host. The URI scheme, client library, and application code stay the same:

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

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.

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 for the full grammar.

Canonical Client Example

The examples below use pika 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)

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

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

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

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)

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)

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

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:

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

See Authentication & security for JWT issuance and Casbin policy syntax.

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)

These AMQP methods are rejected immediately — your application will receive a protocol-level error and must stop using them:

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)

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 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 and Error codes for the exhaustive per-method error tables and wire-contract detail.

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.

   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.

   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