Migrating from Other Transports

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

• An existing MassTransit application currently using UsingRabbitMq, UsingAzureServiceBus, or UsingAmazonSqs
• A KubeMQ broker to migrate onto (see Start KubeMQ below)

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<T> 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<T>, AddSaga<T>, 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

If you do not already have a broker, run KubeMQ in Docker with the gRPC port exposed for the transport:

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

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:

dotnet add package MassTransit.KubeMQ

The transport targets net8.0 and requires MassTransit >= 8.5.0.

From RabbitMQ

Swap UsingRabbitMq for UsingKubeMQ. The RabbitMQ host took a username and password; the KubeMQ host takes an optional AuthToken instead.

services.AddMassTransit(x =>
{
    x.AddConsumer<OrderConsumer>();

    x.UsingRabbitMq((ctx, cfg) =>
    {
        cfg.Host("rabbitmq://localhost", h =>
        {
            h.Username("guest");
            h.Password("guest");
        });
        cfg.ConfigureEndpoints(ctx);
    });
});

services.AddMassTransit(x =>
{
    x.AddConsumer<OrderConsumer>();

    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

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.

services.AddMassTransit(x =>
{
    x.AddConsumer<OrderConsumer>();

    x.UsingAzureServiceBus((ctx, cfg) =>
    {
        cfg.Host("sb://my-namespace.servicebus.windows.net");
        cfg.ConfigureEndpoints(ctx);
    });
});

services.AddMassTransit(x =>
{
    x.AddConsumer<OrderConsumer>();

    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

From Amazon SQS/SNS

Swap UsingAmazonSqs for UsingKubeMQ. The SQS host took a region plus access/secret keys; KubeMQ uses a host, port, and optional auth token.

services.AddMassTransit(x =>
{
    x.AddConsumer<OrderConsumer>();

    x.UsingAmazonSqs((ctx, cfg) =>
    {
        cfg.Host("us-east-1", h =>
        {
            h.AccessKey("my-access-key");
            h.SecretKey("my-secret-key");
        });
        cfg.ConfigureEndpoints(ctx);
    });
});

services.AddMassTransit(x =>
{
    x.AddConsumer<OrderConsumer>();

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

dotnet add package MassTransit.KubeMQ

x.UsingKubeMQ((context, cfg) =>
{
    cfg.Host("localhost", 50000);
    cfg.UseEventsStore();  // all Publish<T>() calls now use EventsStore

    // KubeMQ does not support ConfigureEndpoints() — declare endpoints explicitly.
    cfg.ReceiveEndpoint("order-consumer", e =>
    {
        e.ConfigureKubeMQ(t => { });
    });
});

Decision Guidance After Migrating

Once the swap compiles and runs, two decisions determine how your messaging behaves on KubeMQ.

Events vs EventsStore

Publish<T>() 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.

Enable it globally with cfg.UseEventsStore() or per endpoint with e.UseEventsStore(). See the Publish → EventsStore page for start positions (StartFromFirst, StartFromLast, StartFromSequence, StartFromTime, StartFromTimeDelta).

Queries vs Commands

Request/response maps to KubeMQ CQ. Choose the mode by what the response needs to carry.

See the Commands & Queries page for configuring CqMode globally and per endpoint.

Delayed Delivery Differences

Each transport expresses scheduled delivery differently, and they all converge on one KubeMQ mechanism:

In MassTransit terms you set the delay on the send context, and the transport maps it to QueueMessage.DelaySeconds:

await endpoint.Send(reminder, context =>
{
    context.Delay = TimeSpan.FromSeconds(5);
}, stoppingToken);

Dead-Letter Mapping

Dead-letter handling also converges. Whatever you used before maps to KubeMQ's native dead-letter 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 guide for inspecting dead-lettered messages and configuring native DLQ behavior.

Related