Error Handling & Dead-Lettering

When a consumer throws or a message can't be delivered, MassTransit doesn't lose it — it routes the message to a side channel so you can inspect, replay, or alert on it. On the KubeMQ transport this maps to a pair of dedicated channels per endpoint, plus an optional native dead-letter queue (DLQ) that the broker manages by receive count. This page covers both, along with the retry pipeline that runs before a message ever reaches _error, message expiration as an error-avoidance tool, the transport exception hierarchy, and a troubleshooting checklist.

Prerequisites

• A MassTransit bus already configured with UsingKubeMQ (or AddKubeMQRider) and at least one ReceiveEndpoint (see Configuration)

Error and Skipped Channels

Every receive endpoint gets two automatically-derived side channels, following KubeMQ's _error / _skipped naming convention:

For an endpoint named order-processing, faulted messages route to order-processing_error and skipped messages to order-processing_skipped. No topology to provision — the channels are created on first use, exactly like any other KubeMQ channel.

A faulted message (after retries) routes to _error; an undeliverable one routes to _skipped.

Fault Metadata Tags

When a message is routed to {channel}_error, the transport attaches fault metadata as KubeMQ tags (surfaced as MassTransit headers). Read them on the error channel to understand what failed:

To inspect faults, register a consumer on the _error channel and read the headers:

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<FaultyOrderConsumer>();
    x.AddConsumer<ErrorChannelConsumer>();
    x.UsingKubeMQ((context, cfg) =>
    {
        cfg.Host("localhost", 50000, h => h.ClientId = "error-channel-demo");

        // Main endpoint -- consumer throws on certain messages
        cfg.ReceiveEndpoint("error-demo-orders", e =>
        {
            e.ConfigureKubeMQ(t => { });
        });

        // Error channel endpoint -- receives faulted messages
        cfg.ReceiveEndpoint("error-demo-orders_error", e =>
        {
            e.ConfigureKubeMQ(t => { });
        });
    });
});

public class ErrorChannelConsumer(ILogger<ErrorChannelConsumer> logger)
    : IConsumer<OrderMessage>
{
    public Task Consume(ConsumeContext<OrderMessage> context)
    {
        logger.LogWarning(
            "Faulted order on error channel: OrderId={OrderId}",
            context.Message.OrderId);

        // Inspect fault metadata from headers/tags
        if (context.Headers.TryGetHeader("MT-Fault-ExceptionType", out var exType))
            logger.LogWarning("  Fault exception type: {ExceptionType}", exType);
        if (context.Headers.TryGetHeader("MT-Fault-Message", out var exMsg))
            logger.LogWarning("  Fault message: {FaultMessage}", exMsg);
        if (context.Headers.TryGetHeader("MT-Fault-Timestamp", out var ts))
            logger.LogWarning("  Fault timestamp: {Timestamp}", ts);

        return Task.CompletedTask;
    }
}

Native DLQ via MaxReceiveCount

Beyond the MassTransit-managed _error channel, the KubeMQ transport exposes a native dead-letter queue backed by the broker's receive-count tracking. Configure it per endpoint with ConfigureKubeMQ:

public interface IKubeMQEndpointTransportConfigurator
{
    /// <summary>Set message expiration in seconds.</summary>
    int ExpirationSeconds { set; }

    /// <summary>Configure KubeMQ native DLQ via MaxReceiveCount.</summary>
    void UseNativeDlq(int maxReceiveCount, string? dlqChannel = null);
}

UseNativeDlq(maxReceiveCount, dlqChannel) tells the broker to move a message to the DLQ channel once it has been received maxReceiveCount times without a successful ack. The dlqChannel argument is optional; pass a name to route dead-lettered messages to a channel of your choosing.

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<DlqOrderConsumer>();
    x.AddConsumer<DlqInspectorConsumer>();
    x.UsingKubeMQ((context, cfg) =>
    {
        cfg.Host("localhost", 50000, h => h.ClientId = "native-dlq-demo");

        // After 3 failed receives, move the message to "orders-dlq"
        cfg.ReceiveEndpoint("dlq-demo-orders", e =>
        {
            e.ConfigureKubeMQ(t => t.UseNativeDlq(3, "orders-dlq"));
        });

        // DLQ inspection endpoint -- reads messages that exceeded max receive count
        cfg.ReceiveEndpoint("orders-dlq", e =>
        {
            e.ConfigureKubeMQ(t => { });
        });
    });
});

The DLQ inspector consumer reads the same fault metadata tags that the error channel carries:

public class DlqInspectorConsumer(ILogger<DlqInspectorConsumer> logger)
    : IConsumer<DlqOrderMessage>
{
    public Task Consume(ConsumeContext<DlqOrderMessage> context)
    {
        logger.LogError(
            "Message reached DLQ after max retries: OrderId={OrderId}",
            context.Message.OrderId);

        if (context.Headers.TryGetHeader("MT-Fault-RetryCount", out var retryCount))
            logger.LogError("  Retry count: {RetryCount}", retryCount);
        if (context.Headers.TryGetHeader("MT-Fault-ExceptionType", out var exType))
            logger.LogError("  Last exception: {ExceptionType}", exType);

        return Task.CompletedTask;
    }
}

This is the same e.ConfigureKubeMQ(k => k.UseNativeDlq(maxReceiveCount, dlqChannel)) shape shown in the receive-endpoint reference. Use the native DLQ when you want the broker — not the application — to enforce a hard ceiling on redelivery attempts.

Error Channel vs Skipped Channel

The two side channels capture different failure modes, and you inspect them the same way: by registering a consumer on the suffixed channel.

cfg.ReceiveEndpoint("error-demo-orders", e =>
{
    e.ConfigureKubeMQ(t => { });
});

cfg.ReceiveEndpoint("error-demo-orders_error", e =>
{
    e.ConfigureKubeMQ(t => { });
});

// Only the skipped-channel consumer is registered -- the main channel
// has no matching consumer, so messages are skipped.
cfg.ReceiveEndpoint("skipped-demo-events", e =>
{
    e.ConfigureKubeMQ(t => { });
});

cfg.ReceiveEndpoint("skipped-demo-events_skipped", e =>
{
    e.ConfigureKubeMQ(t => { });
});

cfg.ReceiveEndpoint("inspection-payments", e =>
{
    e.ConfigureKubeMQ(t => { });
});

// Faulted messages with fault metadata
cfg.ReceiveEndpoint("inspection-payments_error", e =>
{
    e.ConfigureKubeMQ(t => { });
});

// Skipped / undeliverable messages
cfg.ReceiveEndpoint("inspection-payments_skipped", e =>
{
    e.ConfigureKubeMQ(t => { });
});

Retry Pipeline

The MassTransit retry and redelivery pipeline runs before a message is moved to _error — a message faults to the error channel only after all configured retries are exhausted. Retry and redelivery policies are part of the MassTransit middleware pipeline, which is transport-independent: it behaves identically on KubeMQ as it does on any other transport.

cfg.ReceiveEndpoint("retry-demo-queue", e =>
{
    // Native DLQ as the final fallback after retries are exhausted
    e.ConfigureKubeMQ(t => t.UseNativeDlq(5, "retry-demo-dlq"));
});

A common pattern is to combine an application-level retry with exponential backoff and the native DLQ as the terminal fallback. The consumer retries transient failures in-process; if every attempt fails, it re-throws so the transport applies the receive-count DLQ rule:

public class RetryConsumer(ILogger<RetryConsumer> logger)
    : IConsumer<RetryableMessage>
{
    // Exponential backoff intervals (ms)
    private static readonly int[] RetryIntervals = [1000, 2000, 4000];

    public async Task Consume(ConsumeContext<RetryableMessage> context)
    {
        Exception? lastException = null;

        for (var attempt = 0; attempt <= RetryIntervals.Length; attempt++)
        {
            try
            {
                // ... process the message; return on success ...
                return;
            }
            catch (Exception ex) when (attempt < RetryIntervals.Length)
            {
                lastException = ex;
                await Task.Delay(RetryIntervals[attempt], context.CancellationToken);
            }
        }

        // All retries exhausted -- re-throw so the native DLQ catches it
        // after maxReceiveCount.
        if (lastException != null)
            throw lastException;
    }
}

Message Expiration

Expiration is an error-avoidance tool: messages that age out are discarded by the broker rather than being delivered late and failing downstream. Set a per-endpoint TTL via k.ExpirationSeconds, which maps to the underlying QueueMessage.ExpirationSeconds. A message not consumed within the TTL is automatically dropped by KubeMQ.

cfg.ReceiveEndpoint("expiration-demo-queue", e =>
{
    // Messages not consumed within 10 seconds are discarded by KubeMQ
    e.ConfigureKubeMQ(t => t.ExpirationSeconds = 10);
});

You can also set TTL per message at send time. MassTransit's TimeToLive maps to QueueMessage.ExpirationSeconds:

await endpoint.Send(new SubmitOrder { OrderId = "123" }, ctx =>
{
    ctx.TimeToLive = TimeSpan.FromHours(1);
});

Transport Exception Hierarchy

When the transport itself fails (as opposed to a consumer faulting), it raises a KubeMQTransportException or one of its subclasses. All derive from MassTransit's MassTransitException and carry the Channel and ServerAddress involved, when applicable.

A few representative triggers, drawn from the validation and runtime paths:

• KubeMQTransportConfigurationException — Host is invalid: must be non-empty, Port is invalid: {value} must be between 1 and 65535, Delayed delivery is not supported for Events. Use Queues. KubeMQTransportOptions.Validate() runs automatically at bus startup and raises this.
• KubeMQTransportConnectionException — Connection closed when the client connection cannot be re-established, or a gRPC UNAVAILABLE status from the server.
• KubeMQTransportTimeoutException — a send or publish operation that exceeds the configured timeout (check network latency and ConnectionTimeout).
• KubeMQTransportException (general) — Queue send failed: {error}, or Query request failed: {error} / Command request failed: {error} when a CQ request returns Executed = false.

Messages Not Being Received

If consumers go quiet, the cause is usually a pattern or naming mismatch rather than a fault. Work through this checklist:

Debug Logging

When the cause isn't obvious, raise the log levels for the transport and the KubeMQ SDK in appsettings.json:

{
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "MassTransit": "Debug",
      "MassTransit.KubeMQTransport": "Debug",
      "KubeMQ": "Debug"
    }
  }
}

What to look for in the output:

• Connection state changes — ConnectionState transitions in the KubeMQ SDK logs (Idle → Connecting → Ready, or Reconnecting) tell you whether the client is actually connected.
• Poll-loop activity — the queue receive transport logs each poll cycle; empty polls increment the kubemq.transport.poll.empty counter, which is a quick way to confirm the receiver is alive but idle.
• Send / publish durations — the transport records kubemq.transport.send.duration and kubemq.transport.publish.duration, so slow or failing operations show up here before they surface as exceptions.

Related

• Observability — OpenTelemetry metrics and counters, including kubemq.transport.errors for transport-level failure counts and health-check states.
• Error codes — the transport exception hierarchy and the _error / _skipped channel naming rules.
• Concepts — how Send, Publish, and Request/Response map to KubeMQ channels and where dead-letter routing fits.