# Migrating from Other Transports (/integrations/masstransit/how-to/migration) Because the KubeMQ transport plugs in below the MassTransit abstractions, moving an existing application onto KubeMQ is a configuration change rather than a rewrite. **Only the transport registration changes** — the lines inside `AddMassTransit` that select and configure the broker. Everything built on top of those abstractions keeps working unchanged. ## Prerequisites [#prerequisites] * An existing MassTransit application currently using `UsingRabbitMq`, `UsingAzureServiceBus`, or `UsingAmazonSqs` * A KubeMQ broker to migrate onto (see [Start KubeMQ](#start-kubemq) below) ## What Does NOT Change [#what-does-not-change] The following MassTransit features work identically regardless of transport, so they carry over untouched when you migrate to KubeMQ: * **Message contracts** — `IConsumer` and your message types * **Consumer and saga implementations** * **Middleware pipeline** — filters and observers * **Serialization** — JSON, `System.Text.Json`, and other serializers * **Retry and redelivery policies** * **Outbox pattern** — the EF Core outbox is transport-independent * **OpenTelemetry integration** — activity and trace context are propagated via KubeMQ Tags (`MT-TraceParent` / `MT-TraceState`) * **Dependency injection registration** — `AddConsumer`, `AddSaga`, and the rest In practice you swap one `UsingXxx` call for `UsingKubeMQ`, point the host at the KubeMQ broker, and your consumers, sagas, and contracts run as-is. ## Start KubeMQ [#start-kubemq] If you do not already have a broker, run KubeMQ in Docker with the gRPC port exposed for the transport: Port `50000` is the gRPC port the MassTransit.KubeMQ transport connects to. Port `9090` is the shared HTTP server that hosts the REST, CloudEvents, MCP, and A2A connectors. Then replace your transport-specific NuGet package with the KubeMQ transport: ```bash title="terminal" dotnet add package MassTransit.KubeMQ ``` The transport targets `net8.0` and requires MassTransit `>= 8.5.0`. ## From RabbitMQ [#from-rabbitmq] Swap `UsingRabbitMq` for `UsingKubeMQ`. The RabbitMQ host took a username and password; the KubeMQ host takes an optional `AuthToken` instead. ```csharp title="Program.cs — before (RabbitMQ)" services.AddMassTransit(x => { x.AddConsumer(); x.UsingRabbitMq((ctx, cfg) => { cfg.Host("rabbitmq://localhost", h => { h.Username("guest"); h.Password("guest"); }); cfg.ConfigureEndpoints(ctx); }); }); ``` ```csharp title="Program.cs — after (KubeMQ)" services.AddMassTransit(x => { x.AddConsumer(); x.UsingKubeMQ((ctx, cfg) => { cfg.Host("localhost", 50000, h => { h.AuthToken = "my-token"; // replaces username/password }); // KubeMQ does not support ConfigureEndpoints() — declare endpoints explicitly. cfg.ReceiveEndpoint("order-consumer", e => { e.ConfigureKubeMQ(t => { }); }); }); }); ``` ### Key differences [#key-differences] | Feature | RabbitMQ | KubeMQ | | ------------------ | -------------------------------------- | ---------------------------------------------------- | | Authentication | Username/Password | Auth Token | | Exchanges/Bindings | Exchange types, bindings, routing keys | Direct channel naming (auto-created) | | Request/Response | Temporary reply queues | Native CQ pattern (no temp queues) | | Delayed delivery | RabbitMQ delayed exchange plugin | `QueueMessage.DelaySeconds` (queues only) | | Publish semantics | Exchange fan-out | Events (fire-and-forget) or EventsStore (persistent) | | Dead letter | DLX exchange + binding | `_error` / `_skipped` channels | | Priority | Queue priority levels | Separate priority channels with weighted polling | | Topology | Exchanges, queues, bindings | Flat channels (auto-created on use) | There is no exchange topology to recreate. KubeMQ channels are created automatically on first use — there is no equivalent of RabbitMQ exchanges, bindings, or routing keys. Channel names are derived from message types and endpoint names. ## From Azure Service Bus [#from-azure-service-bus] Swap `UsingAzureServiceBus` for `UsingKubeMQ`. Azure Service Bus used a fully-qualified namespace; KubeMQ uses a host, port, and optional auth token. ```csharp title="Program.cs — before (Azure Service Bus)" services.AddMassTransit(x => { x.AddConsumer(); x.UsingAzureServiceBus((ctx, cfg) => { cfg.Host("sb://my-namespace.servicebus.windows.net"); cfg.ConfigureEndpoints(ctx); }); }); ``` ```csharp title="Program.cs — after (KubeMQ)" services.AddMassTransit(x => { x.AddConsumer(); x.UsingKubeMQ((ctx, cfg) => { cfg.Host("kubemq-server.example.com", 50000, h => { h.AuthToken = "my-token"; }); // KubeMQ does not support ConfigureEndpoints() — declare endpoints explicitly. cfg.ReceiveEndpoint("order-consumer", e => { e.ConfigureKubeMQ(t => { }); }); }); }); ``` ### Key differences [#key-differences-1] | Feature | Azure Service Bus | KubeMQ | | -------------------- | ------------------------------------- | ------------------------------------------ | | Protocol | AMQP over TCP/WebSocket | gRPC | | Topics/Subscriptions | Topics with subscriptions and filters | Events/EventsStore channels | | Sessions | Session-based ordering | Channel-level ordering | | Scheduled messages | `ScheduledEnqueueTimeUtc` | `QueueMessage.DelaySeconds` | | Dead letter | Built-in DLQ per queue | `_error` / `_skipped` channels | | Duplicate detection | Built-in | Not built-in (handle at application level) | | Auto-provisioning | Requires pre-creation or auto-create | Channels auto-create on use | KubeMQ has no topic-subscription **filters** and no **sessions**. Events and EventsStore deliver to all subscribers on a channel, so content-based routing must happen at the consumer level, and ordered processing within a partition is achieved with a single consumer per channel rather than a session. ## From Amazon SQS/SNS [#from-amazon-sqssns] Swap `UsingAmazonSqs` for `UsingKubeMQ`. The SQS host took a region plus access/secret keys; KubeMQ uses a host, port, and optional auth token. ```csharp title="Program.cs — before (Amazon SQS)" services.AddMassTransit(x => { x.AddConsumer(); x.UsingAmazonSqs((ctx, cfg) => { cfg.Host("us-east-1", h => { h.AccessKey("my-access-key"); h.SecretKey("my-secret-key"); }); cfg.ConfigureEndpoints(ctx); }); }); ``` ```csharp title="Program.cs — after (KubeMQ)" services.AddMassTransit(x => { x.AddConsumer(); x.UsingKubeMQ((ctx, cfg) => { cfg.Host("kubemq-server.example.com", 50000, h => { h.AuthToken = "my-token"; }); // KubeMQ does not support ConfigureEndpoints() — declare endpoints explicitly. cfg.ReceiveEndpoint("order-consumer", e => { e.ConfigureKubeMQ(t => { }); }); }); }); ``` ### Key differences [#key-differences-2] | Feature | Amazon SQS/SNS | KubeMQ | | ------------------ | ------------------------------------------- | ------------------------------------------- | | Queue type | SQS Standard/FIFO | KubeMQ Queues | | Pub/Sub | SNS Topics → SQS Subscriptions | Events/EventsStore | | Request/Response | Simulated via temporary queues | Native CQ pattern | | Visibility timeout | SQS visibility timeout | Manual ack/nack | | Dead letter | Redrive policy + DLQ | `_error` / `_skipped` channels | | Message delay | SQS `DelaySeconds` (0–900s) | `QueueMessage.DelaySeconds` (no 900s limit) | | FIFO ordering | FIFO queues with dedup | Channel-level ordering | | Batch receive | `ReceiveMessage` with `MaxNumberOfMessages` | `PollAsync` with `MaxMessages` | There is no SNS→SQS wiring to recreate. KubeMQ Events go directly to subscribers without an intermediary like SNS, and there is no separate FIFO queue type or deduplication ID — ordering is provided per channel. KubeMQ uses explicit ack/nack instead of a visibility timeout, so failed messages are nack'd immediately rather than waiting for a timeout to redeliver. ## Key Conceptual Shifts [#key-conceptual-shifts] Beyond the host configuration, a few KubeMQ behaviors differ from the transport you are leaving. Understanding these up front avoids surprises after the swap. * **Channels auto-create — no topology to provision.** KubeMQ channels are created on first use. There are no exchanges, bindings, routing keys, ARM/Bicep templates, or portal configuration. Channel names are derived from message types and endpoint names. * **Publish is fire-and-forget by default.** KubeMQ Events do not persist; if no subscriber is active, the message is dropped. Enable `EventsStore` for durable pub/sub that behaves like RabbitMQ's durable queues with bindings or ASB topic subscriptions. * **Request/Response is native.** KubeMQ's CQ pattern provides built-in request-reply with a built-in reply channel and timeout. No temporary reply queues are created, unlike RabbitMQ and SQS which simulate request/response. * **No virtual hosts and no sessions.** KubeMQ has neither RabbitMQ virtual hosts nor Azure Service Bus sessions. Isolation is achieved through channel naming conventions or separate KubeMQ instances; ordered processing uses a single consumer per channel. * **No `ConfigureEndpoints()` auto-configuration.** Unlike the RabbitMQ, Azure Service Bus, and SQS transports, the KubeMQ transport does **not** support `cfg.ConfigureEndpoints(ctx)` — calling it throws `KubeMQTransportConfigurationException` at bus startup. Replace it with one explicit `cfg.ReceiveEndpoint("channel-name", e => { ... })` per channel your consumers read from, as shown in the migration examples above. ## General Migration Checklist [#general-migration-checklist] **Install the package.** Replace the transport-specific NuGet package with `MassTransit.KubeMQ`. ```bash title="terminal" dotnet add package MassTransit.KubeMQ ``` **Update `Program.cs`.** Change `UsingRabbitMq` / `UsingAzureServiceBus` / `UsingAmazonSqs` to `UsingKubeMQ`. **Update host configuration.** Replace the transport-specific host config with the KubeMQ host, port `50000`, and (if your broker has auth enabled) `h.AuthToken`. **Review publish semantics.** Decide whether you need durable pub/sub. If so, enable `cfg.UseEventsStore()` so every `Publish()` is persisted and replayable instead of fire-and-forget. ```csharp title="Program.cs" x.UsingKubeMQ((context, cfg) => { cfg.Host("localhost", 50000); cfg.UseEventsStore(); // all Publish() calls now use EventsStore // KubeMQ does not support ConfigureEndpoints() — declare endpoints explicitly. cfg.ReceiveEndpoint("order-consumer", e => { e.ConfigureKubeMQ(t => { }); }); }); ``` **Review request/response.** KubeMQ CQ is native — verify the `CqMode` (Queries vs Commands) suits your use case. Queries is the default and returns a response body; Commands acknowledges execution only. Switch globally with `cfg.UseCommandsForRequestResponse()`, or override per endpoint with `e.UseCommands()` / `e.UseQueries()`. **Update health checks.** KubeMQ health states map to Healthy / Degraded / Unhealthy in the ASP.NET Core health check pipeline. **Test thoroughly.** Message contracts and consumer logic remain unchanged, but verify transport-specific behaviors (delayed delivery, dead-letter routing, publish durability) against KubeMQ. **Update CI/CD.** Ensure a KubeMQ broker is available in your test and staging environments — for example, the `docker run` snippet above, or a KubeMQ deployment in your cluster. ## Decision Guidance After Migrating [#decision-guidance-after-migrating] Once the swap compiles and runs, two decisions determine how your messaging behaves on KubeMQ. ### Events vs EventsStore [#events-vs-eventsstore] `Publish()` maps to KubeMQ Events (fire-and-forget) by default. Choose **EventsStore** when you need persistence, replay, or late-joining subscribers that catch up from history. | Choose | When | | --------------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | | **Events** | Live fan-out is enough; dropping a message when no subscriber is listening is acceptable (metrics, live dashboards, ephemeral notifications). | | **EventsStore** | You need durability and replay — rebuilding read models, populating a new service, or resuming from a checkpoint after a deployment or incident. | Enable it globally with `cfg.UseEventsStore()` or per endpoint with `e.UseEventsStore()`. See the [Publish → EventsStore](/integrations/masstransit/how-to/events-store) page for start positions (`StartFromFirst`, `StartFromLast`, `StartFromSequence`, `StartFromTime`, `StartFromTimeDelta`). ### Queries vs Commands [#queries-vs-commands] Request/response maps to KubeMQ CQ. Choose the mode by what the response needs to carry. | Choose | When | | --------------------- | ------------------------------------------------------------------------------------------------------------------------------- | | **Queries** (default) | The caller needs a response body — read models, lookups, status checks. The request carries data and the response carries data. | | **Commands** | The caller only needs an execution acknowledgment, not a payload. The response is an ack only (`Executed` / `Error`). | See the [Commands & Queries](/integrations/masstransit/how-to/commands-queries) page for configuring `CqMode` globally and per endpoint. ## Delayed Delivery Differences [#delayed-delivery-differences] Each transport expresses scheduled delivery differently, and they all converge on one KubeMQ mechanism: | Source transport | Source mechanism | KubeMQ mapping | | ----------------- | ------------------------- | ------------------------------------------- | | RabbitMQ | Delayed exchange plugin | `QueueMessage.DelaySeconds` | | Azure Service Bus | `ScheduledEnqueueTimeUtc` | `QueueMessage.DelaySeconds` | | Amazon SQS | `DelaySeconds` (0–900s) | `QueueMessage.DelaySeconds` (no 900s limit) | In MassTransit terms you set the delay on the send context, and the transport maps it to `QueueMessage.DelaySeconds`: ```csharp title="Program.cs — delayed Send" await endpoint.Send(reminder, context => { context.Delay = TimeSpan.FromSeconds(5); }, stoppingToken); ``` Delayed delivery only works on **Queues** (`Send`). It is **not** supported for `Publish` — neither Events nor EventsStore. Publishing with a delay throws `KubeMQTransportConfigurationException`. If you relied on delayed fan-out, route those messages through a queue instead. ## Dead-Letter Mapping [#dead-letter-mapping] Dead-letter handling also converges. Whatever you used before maps to KubeMQ's native dead-letter channels: | Source transport | Source mechanism | KubeMQ mapping | | ----------------- | ---------------------- | ------------------------------ | | RabbitMQ | DLX exchange + binding | `_error` / `_skipped` channels | | Azure Service Bus | Built-in DLQ per queue | `_error` / `_skipped` channels | | Amazon SQS | Redrive policy + DLQ | `_error` / `_skipped` channels | Faulted messages route to `{channel}_error` and skipped messages to `{channel}_skipped`. These channels auto-create like any other — there is no DLX, redrive policy, or DLQ to configure. See the [Error Handling & DLQ](/integrations/masstransit/how-to/error-handling-dlq) guide for inspecting dead-lettered messages and configuring native DLQ behavior. ## Related [#related]