Tributary.Kafka.Producer 1.0.1

Tributary.Kafka.Producer

A .NET Kafka producer with sensible defaults. Wraps Akka.NET Streams Kafka under an opinionated, config-driven API.

dotnet add package Tributary.Kafka.Producer

Bootstrap

builder.Services.AddSharedKafkaProducer(builder.Configuration);

Publish

public sealed class OrdersService
{
    private readonly IEventProducer _producer;

    public OrdersService(IEventProducer producer) => _producer = producer;

    public Task PublishAsync(OrderPlacedEvent evt, CancellationToken ct) =>
        _producer.PublishAsync(
            topic: "orders.placed",
            message: evt,
            key: evt.OrderId,
            options: new PublishOptions
            {
                TenantId = "tenant_001",
                EventType = "OrderPlaced"
            },
            cancellationToken: ct);
}

Every message is wrapped in an EventEnvelope<T> with auto-generated EventId, EventType, OccurredAt, and optional TenantId. The same fields are also attached as Kafka headers (event-id, event-type, occurred-at, tenant-id).

Config

{
  "Kafka": {
    "BootstrapServers": "your-broker:9092",
    "SecurityProtocol": "SASL_SSL",
    "SaslMechanism": "SCRAM-SHA-256",
    "SaslUsername": "your-user",
    "SaslPassword": "your-password",
    "ClientId": "my-service-producer",
    "EnableIdempotence": true
  }
}

The producer applies only the properties you set, so the same code works against Aiven, AWS MSK, Confluent Cloud, or self-hosted Kafka — just change config.

For mTLS (e.g. Aiven Client Certificate auth), use SslCaLocation, SslCertificateLocation, SslKeyLocation instead of the SASL fields.

Idempotent producer (default on)

EnableIdempotence defaults to true, which automatically sets enable.idempotence=true and acks=all. This gives you exactly-once delivery at the producer→broker boundary (no duplicate writes on network retries) for negligible cost. Set to false only if you have a specific reason (legacy broker, throughput-critical fire-and-forget).

Failure handling

PublishAsync throws Confluent.Kafka.ProduceException on failure. The SDK does not swallow errors — your caller decides whether to retry, log, dead-letter, or fail the upstream request.

Common error codes:

Pair with the consumer

See Tributary.Kafka.Consumer for the matching consumer-side SDK with attribute-driven topic binding, configurable commit mode, and optional idempotency.

License

MIT