# Migrating from AMQP 1.0 (/connectors/how-to/migration/from-amqp-1-0) If you have an application that speaks native **AMQP 1.0 (ISO/IEC 19464)** — using a client such as Azure/go-amqp, AMQP.NET Lite, or Apache Qpid Proton — you can point it at KubeMQ's built-in AMQP 1.0 connector by changing **only the endpoint** and, where required, the SASL credentials. Address prefixes select the KubeMQ messaging pattern, so simple publish/consume needs no app-code rewrite. This is a drop-in migration at the **endpoint / client** level. If you are migrating a **JMS** application (Java) instead, see the [Migrating from JMS](/connectors/how-to/migration/from-jms) guide; for **ActiveMQ** applications routed by client type, see [Migrating from ActiveMQ](/connectors/how-to/migration/from-activemq). ## Overview [#overview] The AMQP 1.0 connector exposes KubeMQ's Queues, Events, Events Store, Commands, and Queries patterns over the native AMQP 1.0 wire protocol. It listens on the same ports as the AMQP 0-9-1 connector — **5672** (plain / SASL) and **5671** (TLS / mTLS) — because both protocols share a single listener that routes each connection by its protocol header. No separate firewall rule is needed beyond what the AMQP port already allows. * **Canonical client (this guide):** `github.com/Azure/go-amqp` **v1.7.0**. For .NET shops, **AMQP.NET Lite** is a direct alternative; the connection-string and address conventions are the same, but the snippets below target `go-amqp`. * **Opt-in default:** The connector is **disabled by default**. Set `CONNECTORS_AMQP10_ENABLE=true` (or `Enable = true` under `[Connectors.Amqp10]` in TOML) before connecting. Enable the connector before you migrate any traffic: The enable variable is **`CONNECTORS_AMQP10_ENABLE`** — the literal `10` stays attached to `AMQP` with no underscore. `CONNECTORS_AMQP_1_0_ENABLE` and `CONNECTORS_AMQP10ENABLE` do **not** bind. For Kubernetes, set `spec.amqp10.enabled: true` in the `KubemqCluster` CR. For the full wire-protocol contract, see the [AMQP connector capabilities reference](/connectors/amqp/reference/capabilities). ## Compatibility Matrix [#compatibility-matrix] The cells below describe the AMQP 1.0 column of the cross-protocol migration matrix. | Dimension | Support | Notes | | -------------------------------------- | -------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | **Drop-in level** | endpoint / client | Change the host in the AMQP URI; prefix addresses with the pattern. No app-code rewrite for simple publish/consume. | | **Point-to-point queues** | ✅ | `/queues/` → KubeMQ Queues; competing consumers, ack/nack, visibility. | | **Pub/sub (non-durable)** | ✅ | `/events/` → Events; fire-hose fan-out, sender-settled (at-most-once). | | **Durable / persistent subscriptions** | ✅ | `/events-store/` → Events Store; backed by the persistence engine, resume from last acked position. | | **Request / reply (RPC)** | ✅ | `/commands/` / `/queries/` → Commands / Queries; hand-rolled reply receiver (see [snippet](#canonical-client-example)). | | **Ordering guarantee** | ⚠️ node-local | Within one node; ordering is not cluster-wide. | | **Transactions** | ❌ | AMQP `coordinator` / `declare` / `discharge` frames are not implemented. | | **Dead-letter / redrive** | ❌ no client DLQ | No client-settable DLQ over this protocol. Poison messages that exceed `MaxReceiveCount` are **silently dropped** by the broker, not delivered to a dead-letter address.¹ | | **Selectors / filtering** | ✅ selectors (SQL92 subset) | `apache.org:selector-filter:string` on Events / Events Store links; not supported on `/queues/` links. | | **Auth model** | PLAIN (JWT) / EXTERNAL | SASL PLAIN: password = KubeMQ JWT. SASL EXTERNAL: mTLS, cert CN = ClientID. | | **TLS / mTLS** | ✅ 5671 | Active when the top-level `Security` block is configured. | | **Top unsupported** | transactions; durable-unsub node-local | See [What Does Not Migrate](#what-does-not-migrate--deviations). | ¹ There is no client-settable DLQ over this protocol. The AMQP 1.0 connector never marks published messages for dead-lettering, so a poison message that exceeds `MaxReceiveCount` is silently dropped by the broker rather than delivered to any consumable dead-letter address. For a genuine client-facing DLQ, use the RabbitMQ (DLX) or AWS (redrive) connector. ## Connection / Endpoint Migration [#connection--endpoint-migration] Change only the host and, if required, the SASL credentials. The AMQP 1.0 port is shared with AMQP 0-9-1 on the same listener, so no firewall change is needed beyond what the AMQP port already allows. | | Before (existing broker) | After (KubeMQ) | | --------- | --------------------------- | ------------------------------------- | | **Plain** | `amqp://broker:5672` | `amqp://kubemq-host:5672` | | **TLS** | `amqps://broker:5671` | `amqps://kubemq-host:5671` | | **Auth** | Broker-specific credentials | SASL PLAIN, password = **KubeMQ JWT** | ```go // go-amqp v1.7.0 — amqp.Dial conn, err := amqp.Dial(ctx, "amqp://kubemq-host:5672", &amqp.ConnOptions{ SASLType: amqp.SASLTypePlain("svc-orders", ""), }) ``` When `Authentication.Enable = false` on the server, the connector also accepts SASL ANONYMOUS and bare AMQP headers (no SASL) — convenient for local development. ## Concept & Destination Mapping [#concept--destination-mapping] The **address prefix** of the AMQP link selects the KubeMQ messaging pattern. The leading `/` is optional (`queues/orders` ≡ `/queues/orders`). | Source concept | AMQP 1.0 address | KubeMQ pattern | Channel name | | ------------------------------ | ------------------------------------ | ----------------- | ----------------- | | Queue / P2P | `/queues/` | **Queues** | `` | | Topic / pub-sub | `/events/` | **Events** | `` | | Durable topic / persistent sub | `/events-store/` | **Events Store** | `` | | Command (fire-and-forget RPC) | `/commands/` | **Commands** | `` | | Query (request-response RPC) | `/queries/` | **Queries** | `` | | RPC reply token | `/responses/` | reply path | connection-scoped | | Temporary / dynamic node | `source.dynamic` or `target.dynamic` | in-memory mailbox | node-local | **Selectors (Events / Events Store only):** attach a filter under the `apache.org:selector-filter:string` descriptor on the receiver link source. The SQL92 subset supported includes comparisons, `AND` / `OR` / `NOT`, `BETWEEN`, `IN`, `LIKE`, `IS NULL`, and parentheses, evaluated against application-properties (and standard JMS headers). **Bare addresses:** when no prefix is present the connector resolves by JMS terminus capability hint (`queue` → Queues, `topic` → Events) or falls back to `DefaultPattern` (default: `"queues"`). **Interop with AMQP 0-9-1:** channels produced over the AMQP 0-9-1 connector use the prefix `amqp..`; an AMQP 1.0 client reaches the same data at `/queues/amqp..`. See the [address mapping reference](/connectors/amqp/reference/address-mapping) for the full grammar and the longest-prefix rule. ## From other AMQP 1.0 brokers (Solace / Azure Service Bus) [#from-other-amqp-10-brokers-solace--azure-service-bus] The connector speaks **standard AMQP 1.0**, so non-ActiveMQ AMQP 1.0 clients migrate the same way — change the endpoint, then map destinations to `/`. * **Solace PubSub+** — a Solace AMQP 1.0 sender/receiver targets a queue or topic by name. Remap the Solace destination to `queues/` (persistent) or `events/` (direct). Solace exclusive/non-exclusive durable topic endpoints map to `events-store/` durable subscriptions. Solace selectors map onto the pub/sub selector (`events/`-only). * **Azure Service Bus** — Service Bus AMQP 1.0 entities (`queues/`, `topics//subscriptions/`) remap to `queues/` and `events-store/` (durable). Azure SB sessions, scheduled/deferred delivery, dead-lettering, and transactions have **no KubeMQ equivalent** — drop those features (see [Capabilities](/connectors/amqp/reference/capabilities)). Azure SB's `amqps://` + SAS-token auth maps to KubeMQ SASL PLAIN with a JWT. For any AMQP 1.0 broker, the discipline is identical: **explicit `/` addresses**, continuous credit for at-most-once patterns, symbolic `amqp:*` error conditions (never numeric codes), and the deviations below. ## Canonical Client Example [#canonical-client-example] > **Client:** `github.com/Azure/go-amqp` **v1.7.0** > **Symbols used:** `amqp.Dial`, `conn.NewSession`, `session.NewSender`, `session.NewReceiver`, > `sender.Send`, `amqp.NewMessage`, `receiver.Receive`, `receiver.AcceptMessage`, plus > `msg.Properties.ReplyTo` / `msg.Properties.CorrelationID` for RPC. ### Publish and consume (Queues) [#publish-and-consume-queues] ```go package main import ( "context" "fmt" "log" amqp "github.com/Azure/go-amqp" // v1.7.0 ) func main() { ctx := context.Background() // amqp.Dial establishes the TCP connection and SASL handshake. conn, err := amqp.Dial(ctx, "amqp://kubemq-host:5672", &amqp.ConnOptions{ SASLType: amqp.SASLTypePlain("svc-orders", ""), }) if err != nil { log.Fatal(err) } defer conn.Close() // conn.NewSession opens an AMQP session. sess, err := conn.NewSession(ctx, nil) if err != nil { log.Fatal(err) } // --- Publish --- // session.NewSender attaches a sender link to /queues/orders. snd, err := sess.NewSender(ctx, "/queues/orders", nil) if err != nil { log.Fatal(err) } // sender.Send transfers one message; amqp.NewMessage wraps the body. if err := snd.Send(ctx, amqp.NewMessage([]byte(`{"id":"1","item":"widget"}`)), nil); err != nil { log.Fatal(err) } snd.Close(ctx) // --- Consume --- // session.NewReceiver attaches a competing-consumer receiver on the same queue. rcv, err := sess.NewReceiver(ctx, "/queues/orders", &amqp.ReceiverOptions{ Credit: 10, // grant initial link credit }) if err != nil { log.Fatal(err) } // receiver.Receive blocks until a message arrives. msg, err := rcv.Receive(ctx, nil) if err != nil { log.Fatal(err) } fmt.Printf("received: %s\n", msg.GetData()) // receiver.AcceptMessage settles the delivery (DISPOSITION accepted → broker AckRange). if err := rcv.AcceptMessage(ctx, msg); err != nil { log.Fatal(err) } rcv.Close(ctx) } ``` ### Events Store (durable subscription) [#events-store-durable-subscription] ```go // session.NewReceiver on /events-store/ → durable Events Store subscription. // The broker resumes from the last acknowledged position on reconnect. rcv, err := sess.NewReceiver(ctx, "/events-store/audit", &amqp.ReceiverOptions{ Credit: 64, Durability: amqp.DurabilityUnsettledState, // terminus expiry-policy 'never' }) ``` ### Selectors (Events / Events Store) [#selectors-events--events-store] ```go // Attach a SQL92 selector on an events receiver link source filter. // Selector is evaluated in the connector before delivery. rcv, err := sess.NewReceiver(ctx, "/events/orders", &amqp.ReceiverOptions{ Credit: 32, Filters: []amqp.LinkFilter{ amqp.NewSelectorFilter("priority > 5 AND region = 'EU'"), }, }) ``` ### RPC — hand-rolled reply receiver [#rpc--hand-rolled-reply-receiver] The AMQP 1.0 connector has **no library-level request/reply helper**. You must create a dynamic reply receiver yourself, set `msg.Properties.ReplyTo` to its address, and match responses by `CorrelationID`. The example below uses the exact `go-amqp v1.7.0` symbols. ```go // 1. Open a dynamic receiver to serve as the reply address. // session.NewReceiver with DynamicAddress=true → connector allocates // a temporary node and returns its address in the ATTACH reply. replyRcv, err := sess.NewReceiver(ctx, "", &amqp.ReceiverOptions{ Credit: 1, DynamicAddress: true, }) if err != nil { log.Fatal(err) } replyAddr := replyRcv.Address() // the connector-assigned dynamic node address // 2. Attach a sender to the command channel. snd, err := sess.NewSender(ctx, "/commands/status", nil) if err != nil { log.Fatal(err) } // 3. Build the request message. // msg.Properties.ReplyTo tells the connector where to send the response. // msg.Properties.CorrelationID allows matching the reply to the request. req := amqp.NewMessage([]byte(`{"service":"inventory"}`)) req.Properties = &amqp.MessageProperties{ ReplyTo: &replyAddr, CorrelationID: "req-001", } // sender.Send dispatches the request to the Commands channel. if err := snd.Send(ctx, req, nil); err != nil { log.Fatal(err) } // 4. receiver.Receive blocks for the reply; the connector routes it to replyAddr. reply, err := replyRcv.Receive(ctx, nil) if err != nil { log.Fatal(err) } fmt.Printf("reply correlation=%v body=%s\n", reply.Properties.CorrelationID, reply.GetData()) // receiver.AcceptMessage acknowledges the reply delivery. replyRcv.AcceptMessage(ctx, reply) replyRcv.Close(ctx) snd.Close(ctx) ``` ## Security [#security] ### Authentication [#authentication] | Mechanism | When | Credential | | ---------------- | ------------------------------------- | -------------------------------------------------------------- | | `SASL PLAIN` | Always available | password = **KubeMQ JWT**; username is recorded for audit only | | `SASL ANONYMOUS` | `Authentication.Enable = false` | No credentials; ClientID derived from `container-id` | | `SASL EXTERNAL` | mTLS with verified client certificate | Certificate CN becomes the ClientID — no JWT needed | ### TLS / mTLS [#tls--mtls] The TLS listener on port **5671** activates only when the top-level `Security` block is configured. Point clients at `amqps://kubemq-host:5671`. For mutual TLS, configure the server to request client certificates; the certificate CN then serves as the connection ClientID. ### Authorization [#authorization] With `Authorization.Enable = true`, the connection's ClientID is checked against the Casbin policy per link: * **Sender link** (client → KubeMQ): `Write` on the resolved channel, checked at ATTACH. * **Receiver link** (KubeMQ → client): `Read` on the resolved channel, checked at ATTACH. * **Anonymous-terminus sender**: per-message `Write` check against `properties.to` (1024-entry LRU cache, 60 s TTL). * **`/responses/` reply token**: no policy check (connection-scoped). ### Minimal TOML configuration [#minimal-toml-configuration] ```toml title="config.toml" [Connectors.Amqp10] Enable = true Port = 5672 # shared with [Connectors.Amqp] (0-9-1) via the same listener TlsPort = 5671 # active only when [Security] is configured ``` ## What Does NOT Migrate / Deviations [#what-does-not-migrate--deviations] ### Not supported [#not-supported] | Feature | Detail | | ---------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | **AMQP transactions** | The `coordinator`, `declare`, `discharge`, transactional acquisition, and transactional retirement performatives are not implemented. JMS `SESSION_TRANSACTED` sessions do not work; use `AUTO_ACKNOWLEDGE` or `CLIENT_ACKNOWLEDGE`. XA / two-phase commit is not available. | | **`rcv-settle-mode = second`** | Two-phase receiver settlement is not supported (DETACH `amqp:not-implemented`). Use `first` (the go-amqp default). | | **`AmqpSequence` body sections** | Rejected (`amqp:not-implemented`). Use Data sections or `AmqpValue`. | | **AMQP-over-WebSocket** | No WebSocket binding (RFC 7395). Use raw TCP on 5672 / TLS on 5671. | | **SASL SCRAM / GSSAPI / Azure CBS** | Only PLAIN, ANONYMOUS, and EXTERNAL are offered. | | **AMQP management (node create/delete)** | Use the KubeMQ REST API or dashboard instead. | | **Client-settable DLQ / redrive** | No client-settable DLQ over this protocol. See footnote ¹ in the Compatibility Matrix for the behavior and alternative-connector options (RabbitMQ DLX / AWS redrive). | ### Behavioral deviations [#behavioral-deviations] | Area | Behavior | | ------------------------------------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | **Durable-subscription unsubscribe is node-local** | Durable Events Store subscription identities are tracked per node. The same durable subscription attached on two cluster nodes causes a durable-subscription client-ID conflict. An `unsubscribe()` call or durable detach is local to the node that owns the registration; if the client reconnects to a different node the durable state may not transfer cleanly. Retry the attach on conflict. | | **Pub/sub is at-most-once (sender-settled fire-hose)** | `/events/` links: events are dropped when link credit is 0 (`kubemq_amqp10_events_dropped_no_credit_total`), not buffered. Grant credit continuously for a durable-enough consumer. | | **Events Store stalled-credit link detach** | A bounded per-link buffer (`MaxUnsettledPerLink`) fronts the Events Store subscription. If the buffer fills while credit stays at 0, the link is detached (`amqp:resource-limit-exceeded`) and the buffered window is dropped. Affected positions are already acked, so a durable re-attach resumes *after* them. Size `MaxUnsettledPerLink` to match the consumer's expected burst. | | **`released` increments receive count** | The broker increments `ReceiveCount` on redelivery after a `released` / `modified{delivery-failed=true}` settlement. This counts toward `MaxReceiveCount` — a strict AMQP reading would not count a release as a delivery attempt. | | **Selectors on `/queues/` links** | Rejected (`amqp:not-implemented`). Selectors work only on Events and Events Store receivers. | | **Dynamic (temporary) nodes are node-local** | A temp reply node lives in memory on the owning node only. Direct cross-connection sends to another connection's temp node work only within the same node. RPC replies that travel through the broker path (`/responses/`) are unaffected. | | **`header.priority` does not schedule** | The field round-trips as the `amqp10.priority` tag but drives no priority ordering. | | **Config hot-reload** | Changing `Connectors.Amqp10.*` requires a server restart. | ## Verification Smoke Test [#verification-smoke-test] This recipe uses the [publish and consume snippet](#canonical-client-example) above as a copy-pasteable confirmation that the migration is working. **Prerequisites:** 1. KubeMQ is running with `Connectors.Amqp10.Enable = true`. 2. A KubeMQ JWT is available (or `Authentication.Enable = false` for local testing). **Steps:** ```go // Smoke test — publish one message to /queues/smoke-test, consume it, confirm arrival. // Run the main() from the Queues snippet above with destination "/queues/smoke-test". // Expected output: received: ``` 1. Run the snippet targeting `/queues/smoke-test`. 2. Confirm `received: ...` appears in stdout. 3. If the `amqp.Dial` call fails, check that `CONNECTORS_AMQP10_ENABLE=true` is set and port 5672 is reachable. 4. If `sender.Send` times out, verify the JWT in the SASL PLAIN password field is valid. 5. For Events / Events Store, swap the address prefix and confirm fan-out to multiple receivers. 6. For RPC, run the command snippet and confirm a reply with matching `CorrelationID` is received within the `DefaultRpcTimeoutSeconds` window (default: 30 s). ## See Also [#see-also]