Kafka Transport

Production tier: GA (transport platform and dispatch). Kafka ingress is Beta; see v6 feature index.

LiteBus.Transport.Kafka wraps Confluent.Kafka for publish and consume. Inbox and outbox adapters compose it independently; installing Kafka transport does not pull AMQP or AWS SDKs.

Packages to Install

Package	Role
`LiteBus.Transport.Kafka`	`ITransportPublisher`, `KafkaConsumer`, connection options
`LiteBus.Inbox.Dispatch.Kafka`	Outbound command dispatch from inbox processor
`LiteBus.Outbox.Dispatch.Kafka`	Outbound event publish from outbox processor
`LiteBus.Inbox.Ingress.Kafka`	Broker intake into `IInbox.AcceptAsync`

Add inbox/outbox core and storage packages as usual. See Dependency graph.

Registration

var kafkaOptions = new KafkaTransportOptions
{
    BootstrapServers = "localhost:9092",
    ConnectivityCheckTimeout = TimeSpan.FromSeconds(3)
};

builder.AddKafkaTransport(kafkaOptions);
builder.AddMessaging(_ => { });
builder.AddInbox(inbox =>
{
    inbox.Contracts.Register<ShipOrderCommand>("orders.commands.ship", 1);
    inbox.UsePostgreSqlStorage(pg => pg.UseConnectionString(connectionString));
    inbox.UseInProcessDispatch();

    inbox.UseKafkaDispatch(kafka => kafka.DefaultDestination = "orders.commands");

    inbox.UseKafkaIngress(ingress =>
    {
        ingress.UseOptions(new KafkaInboxIngressOptions
        {
            Destination = "orders.commands",
            RequeueOnFailure = true
        });
    });

    inbox.EnableInboxProcessor();
});

Options Reference

Type	Property	Default	Notes
`KafkaTransportOptions`	`BootstrapServers`	required	Broker list
	`ConsumerGroupId`	`litebus-transport`	Consumer group for ingress
	`ConnectivityCheckTimeout`	5 s	Readiness cluster-description request timeout
	`SeekFailureBackoffInitial`	250 ms	Delay before re-read after seek
	`SeekFailureBackoffMax`	30 s	Cap on seek backoff
	`SeekFailureBackoffMultiplier`	2.0	Per-offset failure multiplier
`KafkaInboxIngressOptions`	`RequeueOnFailure`	`true`	Seek back on transient accept failure
	`Safety.MaxInFlightMessages`	32	Provider-neutral LiteBus handler cap

LiteBus does not expose a Kafka PrefetchCount setting. The Confluent.Kafka .NET consume loop returns records one at a time from the client's background fetch queue. Kafka fetch tuning remains in KafkaTransportOptions or the underlying client configuration when a supported option is added.

Guarantees and Non-Guarantees

Guaranteed	Not guaranteed
At-least-once delivery when handlers ack after successful accept/dispatch	Exactly-once side effects
Offset committed only after `TransportMessage.AcceptAsync`	In-session retry without seek on failure
Seek-on-failure rewinds to failed offset before next consume	Ordering across partitions

On transient ingress failure, ReturnToQueueAsync seeks the consumer to the failed offset. The offset is not committed until accept succeeds. Handlers must be idempotent.

Operations

Symptom	Check	Action
Same message reprocessed repeatedly	Consumer lag, seek backoff logs	Fix root accept/dispatch failure; verify idempotency
Consumer stuck on one offset	Repeated seek at same partition/offset	Inspect store availability; increase backoff cap if broker pressure
No ingress	Topic name, group id, ACLs	Confirm `Destination` topic exists; reset group if needed
`transport.kafka.connectivity` unhealthy	Bootstrap reachability, TLS/SASL, cluster ACLs	Fix broker access; tune `ConnectivityCheckTimeout` only when network latency requires it

Monitor consumer lag, seek retry rate, and inbox/outbox row counts. See Production runbook.

The root transport registers transport.kafka.connectivity. It uses Confluent's asynchronous cluster-description API with an explicit request timeout.

Tests

Scenario	Location
Seek on `ReturnToQueueAsync`	`LiteBus.Transport.UnitTests`
Ingress transient failure redelivery	`LiteBus.Durable.IntegrationTests` (`Ingress/Kafka/`)
Ingress and dispatch round-trip	`LiteBus.Durable.IntegrationTests` (`Ingress/Kafka/`, `Dispatch/Inbox/Kafka/`, `Dispatch/Outbox/Kafka/`)

AMQP transport
AWS SQS transport
Azure Service Bus transport
Inbox AMQP ingress (ingress patterns shared across brokers)
Reliable messaging
Integration tests

On this page