Configuration

Every MassTransit.KubeMQ application configures the transport in one place: the UsingKubeMQ (or AddKubeMQRider) callback inside AddMassTransit. The connection settings — host, port, authentication, TLS, and poll behavior — are modeled by a single configuration POCO, KubeMQTransportOptions, which you can set in code, parse from a kubemq:// URI, or bind from appsettings.json. This page walks through every option and the three ways to supply it.

Before you connect, start a broker with the gRPC port exposed:

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 transport connects to. Port 9090 is the shared HTTP server that hosts the REST, CloudEvents, MCP, and A2A connectors — it is not used by the transport itself.

Prerequisites

• The MassTransit.KubeMQ package installed (dotnet add package MassTransit.KubeMQ) in a MassTransit >= 8.5.0 project targeting net8.0
• The broker above running and reachable

Transport Options

KubeMQTransportOptions is the central configuration class. It is bindable to IOptions<T> and appsettings.json, and the same property names appear on the host configurator (h.AuthToken, h.UseTls, and so on). The table below lists every property with its default and validation range.

Host and Port (Recommended)

The simplest and recommended way to point the transport at a broker is cfg.Host(host, port) — the Host method on IKubeMQBusFactoryConfigurator. The port defaults to 50000, so for a local broker you can omit it entirely.

services.AddMassTransit(x =>
{
    x.UsingKubeMQ((ctx, cfg) =>
    {
        cfg.Host("kubemq-server.example.com", 50000);

        cfg.ReceiveEndpoint("config-options-queue", e =>
        {
            e.ConfigureKubeMQ(t => { });
        });
    });
});

The KubeMQ transport does not support MassTransit's cfg.ConfigureEndpoints(ctx) auto-configuration — calling it throws KubeMQTransportConfigurationException at bus startup (see Error Handling & DLQ for the full exception model). Declare every receive endpoint explicitly with cfg.ReceiveEndpoint("channel-name", e => { ... }), one per channel your consumers read from, as shown above and throughout this guide.

URI-Based Host

Host has a second overload that takes a Uri. Use the kubemq:// scheme, and pass connection options as query parameters. This form is handy when the connection string comes from configuration or an environment variable.

services.AddMassTransit(x =>
{
    x.UsingKubeMQ((ctx, cfg) =>
    {
        cfg.Host(new Uri("kubemq://kubemq-server:50000?authToken=my-token&tls=true"));

        cfg.ReceiveEndpoint("config-options-queue", e =>
        {
            e.ConfigureKubeMQ(t => { });
        });
    });
});

Two query parameters are parsed from the URI:

A common pattern is to read the URI from an environment variable so the same binary runs against different brokers without recompiling:

var connectionString = Environment.GetEnvironmentVariable("KUBEMQ_CONNECTION_STRING")
    ?? "kubemq://localhost:50000";

cfg.Host(new Uri(connectionString), h =>
{
    h.ClientId = "connstring-client";
});

The optional second argument to Host is a host configurator callback — you can combine the URI form with explicit settings such as ClientId, as above.

Authentication and TLS

For full control, pass a host configurator callback to cfg.Host(host, port, h => { ... }). The callback exposes the connection-level settings: AuthToken, UseTls, the three TLS file paths, and the connection/reconnect timeouts.

cfg.Host("kubemq-server.example.com", 50000, h =>
{
    h.ClientId = "auth-token-client";
    h.AuthToken = "my-secret-token";
});

cfg.Host("kubemq-server.example.com", 50000, h =>
{
    h.UseTls = true;
    h.TlsCertFile = "/certs/client.pem";
});

cfg.Host("kubemq-server.example.com", 50000, h =>
{
    h.UseTls = true;
    h.TlsCertFile = "/certs/client.pem";
    h.TlsKeyFile = "/certs/client-key.pem";
    h.TlsCaFile = "/certs/ca.pem";

    h.ConnectionTimeout = TimeSpan.FromSeconds(30);
    h.ReconnectTimeout = TimeSpan.FromSeconds(120);
});

When UseTls is enabled, TlsCertFile is required; TlsKeyFile and TlsCaFile are needed only for mutual TLS, where the client presents its own certificate and validates the server against a CA. ConnectionTimeout (default 10s) bounds the initial connect, and ReconnectTimeout (default 60s) bounds the wait between reconnection attempts.

appsettings.json and IOptions Binding

KubeMQTransportOptions is designed to bind directly from configuration. Define a KubeMQ section in appsettings.json and bind it with Configure<KubeMQTransportOptions>. Note that ConnectionTimeout and ReconnectTimeout are TimeSpan values, so they use the "hh:mm:ss" string format.

{
  "KubeMQ": {
    "Host": "kubemq-server.example.com",
    "Port": 50000,
    "AuthToken": "my-token",
    "UseTls": false,
    "PollTimeoutSeconds": 10,
    "MaxPollMessages": 64,
    "DefaultCqMode": "Queries",
    "ConnectionTimeout": "00:00:30",
    "ReconnectTimeout": "00:02:00"
  }
}

There are two ways to consume the bound options. The first is to bind the section yourself with Configure<T>, then read the resolved IOptions<KubeMQTransportOptions> inside the UsingKubeMQ callback to configure the host:

// Bind the "KubeMQ" section to KubeMQTransportOptions
builder.Services.Configure<KubeMQTransportOptions>(
    builder.Configuration.GetSection("KubeMQ"));

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<OptionsConsumer>();

    x.UsingKubeMQ((context, cfg) =>
    {
        // IOptions<KubeMQTransportOptions> is resolved from DI
        var options = context.GetRequiredService<IOptions<KubeMQTransportOptions>>().Value;

        cfg.Host(options.Host, options.Port, h =>
        {
            if (options.ClientId != null)
                h.ClientId = options.ClientId;
        });

        cfg.ReceiveEndpoint("config-options-queue", e =>
        {
            e.ConfigureKubeMQ(t => { });
        });
    });
});

The second is the UsingKubeMQ overload that takes a configureOptions callback as its last argument. The callback receives a KubeMQTransportOptions instance you can mutate; the values flow into the transport configuration. This is convenient for setting a few options inline without binding a configuration section:

services.AddMassTransit(x =>
{
    x.UsingKubeMQ((ctx, cfg) =>
    {
        cfg.Host("kubemq-server.example.com", 50000);

        cfg.ReceiveEndpoint("config-options-queue", e =>
        {
            e.ConfigureKubeMQ(t => { });
        });
    }, options =>
    {
        options.MaxPollMessages = 64;
        options.PollTimeoutSeconds = 10;
    });
});

Per-Endpoint Overrides

Poll behavior and KubeMQ-specific transport settings can be tuned per receive endpoint, overriding the global defaults. On the endpoint configurator (IKubeMQReceiveEndpointConfigurator) you can set PollTimeoutSeconds and MaxPollMessages directly, and call ConfigureKubeMQ to reach the transport-level options (IKubeMQEndpointTransportConfigurator): ExpirationSeconds for message TTL and UseNativeDlq for the native dead-letter queue.

x.UsingKubeMQ((context, cfg) =>
{
    cfg.Host("localhost", 50000, h =>
    {
        h.ClientId = "per-endpoint-client";
    });

    // Aggressive polling: short timeout, large batch
    cfg.ReceiveEndpoint("config-fast-poll-queue", e =>
    {
        e.PollTimeoutSeconds = 2;
        e.MaxPollMessages = 128;
        e.ConfigureKubeMQ(t => { });
    });

    // Relaxed polling: longer timeout, smaller batch
    cfg.ReceiveEndpoint("config-slow-poll-queue", e =>
    {
        e.PollTimeoutSeconds = 30;
        e.MaxPollMessages = 8;
        e.ConfigureKubeMQ(t => { });
    });

    // Native DLQ + message expiration
    cfg.ReceiveEndpoint("order-processing", e =>
    {
        e.ConfigureKubeMQ(k =>
        {
            k.ExpirationSeconds = 3600;       // messages expire after 1 hour
            k.UseNativeDlq(5, "orders-dlq");  // max 5 receive attempts, then DLQ
        });
    });
});

UseNativeDlq(maxReceiveCount, dlqChannel) configures KubeMQ's native dead-letter queue: after maxReceiveCount failed receive attempts, the message is routed to the named DLQ channel. PollTimeoutSeconds is the long-poll wait — a higher value reduces empty polls on idle queues, while a lower value makes the endpoint more responsive.

Validation

KubeMQTransportOptions.Validate() runs automatically during bus startup. If any value is out of range it throws KubeMQTransportConfigurationException (see Error Handling & DLQ for the full exception model), so misconfiguration fails fast at boot rather than surfacing as an obscure runtime error. The following conditions are rejected:

You can run the same validation yourself before startup — useful in tests or a configuration smoke check:

using MassTransit.KubeMQTransport;
using MassTransit.KubeMQTransport.Exceptions;

try
{
    var options = new KubeMQTransportOptions { Port = 99999 };
    options.Validate();
}
catch (KubeMQTransportConfigurationException ex)
{
    // "Port is invalid: 99999 must be between 1 and 65535"
    Console.WriteLine(ex.Message);
}

Endpoint Address Format

Send endpoints are addressed with the kubemq:// URI scheme. The full form names the host, port, and channel:

kubemq://host:port/channel-name

When resolving a send endpoint at runtime, build the URI from your configured host and port:

var endpoint = await bus.GetSendEndpoint(
    new Uri("kubemq://localhost:50000/order-processing"));

MassTransit also accepts the queue: shorthand, which resolves to a channel on the configured host without repeating host and port:

var endpoint = await bus.GetSendEndpoint(new Uri("queue:order-processing"));

Multi-Transport and Multi-Bus Setups

When the bus already uses another transport — RabbitMQ, Azure Service Bus, or InMemory — and you want to add KubeMQ alongside it, use AddKubeMQRider instead of UsingKubeMQ. The rider configurator (IKubeMQRiderConfigurator) exposes the same Host and ReceiveEndpoint methods, so configuration looks identical.

builder.Services.AddMassTransit(x =>
{
    // Base bus uses InMemory (or any other transport)
    x.UsingInMemory();

    // Add KubeMQ as a supplementary transport
    x.AddKubeMQRider((ctx, k) =>
    {
        k.Host("localhost", 50000);

        k.ReceiveEndpoint("rider-basic-queue", e =>
        {
            e.ConfigureKubeMQ(t => { });
        });
    });
});

This is the foundation for multi-bus and domain-separation topologies, where different services or message domains map to distinct endpoints and channels on the same broker.

Related

• Reference for the complete KubeMQTransportOptions and validation table
• Queues (Send) for queue-specific endpoint settings such as poll behavior and DLQ
• Events Store (Durable Publish) for persistent publish and subscription start positions
• Commands & Queries for CQ-mode (Commands vs Queries) configuration