Producer Options

Complete reference for all producer configuration options.

These methods are available anywhere a ProducerBuilder<TKey,TValue> is used, including dependency injection registration:

// Before: DI examples only showed the small common subset.
builder.Services.AddDekaf(dekaf =>
{
    dekaf.AddProducer<string, string>(producer => producer
        .WithBootstrapServers("localhost:9092"));
});

// After: DI uses the full producer builder.
builder.Services.AddDekaf(dekaf =>
{
    dekaf.AddProducer<string, string>(producer => producer
        .WithBootstrapServers("localhost:9092")
        .WithLinger(TimeSpan.FromMilliseconds(5))
        .WithBatchSize(64 * 1024)
        .WithBufferMemory(256 * 1024 * 1024)
        .WithSaslScramSha512("user", "password")
        .ForHighThroughput());
});

Configuration Binding

Dekaf.Extensions.DependencyInjection can bind producer settings from an IConfiguration section. Keys use ProducerOptions property names:

{
  "Kafka": {
    "Producers": {
      "Orders": {
        "BootstrapServers": "broker1:9092,broker2:9092",
        "ClientId": "orders-producer",
        "Acks": "All",
        "LingerMs": 5,
        "BatchSize": 65536,
        "CompressionType": "Lz4",
        "EnableIdempotence": true,
        "UseTls": true,
        "SaslMechanism": "ScramSha512",
        "SaslUsername": "user",
        "SaslPassword": "password"
      }
    }
  }
}

builder.Services.AddDekaf(dekaf =>
{
    dekaf.AddProducer<string, Order>(
        builder.Configuration.GetSection("Kafka:Producers:Orders"),
        producer => producer.WithValueSerializer(new JsonSerializer<Order>()));
});

Configuration is applied before the optional fluent callback, so fluent calls can override values from appsettings.json.

Fluent API	Config key	Notes
`WithBootstrapServers(...)`	`BootstrapServers`	Server list (prefer `params string[]` in code; comma-separated string and JSON arrays are also supported)
`WithClientId(...)`	`ClientId`	String
`WithClientDnsLookup(...)`	`ClientDnsLookup`	`UseAllDnsIps` or `ResolveCanonicalBootstrapServersOnly`
`WithBootstrapResolveTimeout(...)`	`BootstrapResolveTimeoutMs`	Milliseconds; default 120000
`WithAcks(...)`	`Acks`	`None`, `Leader`, `All`
`WithLinger(...)`	`LingerMs`	Milliseconds
`WithBatchSize(...)`	`BatchSize`	Bytes
`WithBufferMemory(...)`	`BufferMemory`	Bytes; omit to keep auto-tuning
`WithBufferMemoryAllocationStrategy(...)`	`BufferMemoryAllocationStrategy`	`Full` or `Incremental`
`WithMaxBlock(...)`	`MaxBlockMs`	Milliseconds
`WithDeliveryLatencyTarget(...)`	`DeliveryLatencyTargetMs`	TimeSpan; `TimeSpan.Zero` disables
`WithDeliveryTimeout(...)`	`DeliveryTimeoutMs`	Milliseconds
`WithRequestTimeout(...)`	`RequestTimeoutMs`	Milliseconds
`WithIdempotence(...)`	`EnableIdempotence`	Boolean
`WithConnectionsMaxIdle(...)`	`ConnectionsMaxIdleMs`	Milliseconds; `-1` disables
`WithConnectionTimeout(...)`	`ConnectionTimeout`	TimeSpan
`WithConnectionTimeoutMax(...)`	`ConnectionTimeoutMax`	TimeSpan; Kafka `socket.connection.setup.timeout.max.ms`
`WithTcpKeepAlive(...)`	`EnableTcpKeepAlive`, `TcpKeepAliveTime`, `TcpKeepAliveInterval`, `TcpKeepAliveRetryCount`	Socket keepalive
`WithConnectionsPerBroker(...)`	`ConnectionsPerBroker`	Integer
`WithAdaptiveConnections(...)`	`EnableAdaptiveConnections`, `MaxConnectionsPerBroker`	Set `EnableAdaptiveConnections` to `false` to disable
`WithTransactionalId(...)`	`TransactionalId`	String
`WithTransactionTimeout(...)`	`TransactionTimeoutMs`	Milliseconds
`UseCompression(...)`	`CompressionType`	`None`, `Gzip`, `Snappy`, `Lz4`, `Zstd`
`WithCompressionLevel(...)`	`CompressionLevel`	Codec-specific integer
`WithPartitioner(...)`	`Partitioner`	`Default`, `Sticky`, `RoundRobin`
`WithAdaptivePartitioning(...)`	`EnableAdaptivePartitioning`	Boolean; Kafka `partitioner.adaptive.partitioning.enable`
`WithPartitionerAvailabilityTimeout(...)`	`PartitionerAvailabilityTimeoutMs`	Milliseconds; Kafka `partitioner.availability.timeout.ms`
`WithPartitionerIgnoreKeys(...)`	`IgnorePartitionerKeys`	Boolean; Kafka `partitioner.ignore.keys`
`WithClientRack(...)`	`ClientRack`	String; Kafka `client.rack`
`WithRackAwarePartitioning(...)`	`EnableRackAwarePartitioning`	Boolean; Kafka `partitioner.rack.aware`
`UseTls(...)`	`UseTls`, `TlsConfig`	`TlsConfig` can bind certificate path fields
`WithRemoteCertificateValidationCallback(...)`	Runtime callback	Custom TLS certificate validation
`WithSaslPlain(...)` / `WithSaslScramSha512(...)`	`SaslMechanism`, `SaslUsername`, `SaslPassword`	`SaslMechanism` values match the enum names
`WithGssapi(...)`	`SaslMechanism`, `GssapiConfig`	Use `SaslMechanism: Gssapi`
`WithOAuthBearer(...)`	`SaslMechanism`, `OAuthBearerConfig`	Use `SaslMechanism: OAuthBearer`
`WithOAuthBearerJwtBearer(...)`	Runtime callback	Signs JWT assertions with RSA/ECDSA keys
`WithSocketSendBufferBytes(...)`	`SocketSendBufferBytes`	Bytes
`WithSocketReceiveBufferBytes(...)`	`SocketReceiveBufferBytes`	Bytes
`WithMetadataRecoveryStrategy(...)`	`MetadataRecoveryStrategy`	`None` or `Rebootstrap`
`WithMetadataClusterCheck(...)`	`MetadataClusterCheckEnabled`	KIP-1242 identity check; default `true`, ignored with `None` recovery
`WithMetadataRecoveryRebootstrapTrigger(...)`	`MetadataRecoveryRebootstrapTriggerMs`	Milliseconds

Serializers, custom partitioners, interceptors, and retry policies are objects, so configure those in the fluent callback.

Connection Settings

WithBootstrapServers

Kafka broker addresses for initial connection. Prefer the typed params string[] overload in code; the single-string overload remains a convenience for configuration-style comma-separated values.

// Single server
.WithBootstrapServers("localhost:9092")

// Multiple servers (typed params)
.WithBootstrapServers("broker1:9092", "broker2:9092", "broker3:9092")

// Convenience for comma-separated configuration values
.WithBootstrapServers("broker1:9092,broker2:9092,broker3:9092")

WithClientId

Identifier sent to brokers for logging and metrics:

.WithClientId("order-service-producer")

Delivery Settings

WithAcks

Controls when the broker considers a message delivered:

.WithAcks(Acks.All)     // Wait for all in-sync replicas (safest)
.WithAcks(Acks.Leader)  // Wait for leader only (faster)
.WithAcks(Acks.None)    // Don't wait (fastest, may lose messages)

WithIdempotence

Prevents duplicate messages during retries:

.WithIdempotence(true)

Automatically sets Acks.All and enables sequence numbers.

Batching Settings

WithLinger

Time to wait before sending a batch:

.WithLinger(TimeSpan.FromMilliseconds(5))  // Same, using TimeSpan

Higher values = more batching, higher latency.

WithBatchSize

Maximum batch size in bytes:

.WithBatchSize(65536)  // 64KB batches

Compression

UseCompression / Specific Methods

Enable message compression:

.UseLz4Compression()    // Fast, good ratio (recommended)
.UseZstdCompression()   // Best ratio, more CPU
.UseSnappyCompression() // Very fast, lower ratio
.UseGzipCompression()   // Compatible, slower

// Or specify directly
.UseCompression(CompressionType.Lz4)

Partitioning

WithPartitioner

Control how messages are assigned to partitions:

.WithPartitioner(PartitionerType.Default)     // Hash key or sticky null keys
.WithPartitioner(PartitionerType.Sticky)      // Stick null keys for batching
.WithPartitioner(PartitionerType.RoundRobin)  // Even distribution
.WithPartitioner(PartitionerType.ConsistentRandom) // librdkafka default
.WithPartitioner(PartitionerType.Fnv1ARandom)      // librdkafka/Sarama-compatible

The built-in default and sticky partitioners use KIP-794 behavior for sticky records: they stay on a partition until at least BatchSize bytes have been produced to it. Adaptive partitioning is enabled by default and weights new sticky choices away from partitions with queued batches. Set .WithAdaptivePartitioning(false) for uniform switching, .WithPartitionerAvailabilityTimeout(...) to exclude backed-up partitions after a timeout, or .WithPartitionerIgnoreKeys() to use sticky partitioning even when records have keys. KIP-1123 rack-aware partitioning is opt-in with .WithClientRack("rack-a").WithRackAwarePartitioning(). It prefers local partition leaders and falls back to all leaders when none are usable; uneven rack placement can therefore produce an uneven partition distribution.

Transactions

WithTransactionalId

Enable transactional producer:

.WithTransactionalId("my-service-tx-1")

Must be unique per producer instance.

WithTwoPhaseCommit

Enable KIP-939 two-phase commit participation for transactional producers:

.WithTransactionalId("my-service-tx-1")
.WithTwoPhaseCommit()

Requires broker support for transaction.version 3 and InitProducerId v6.

Security

UseTls

Enable TLS encryption:

.UseTls()                    // Basic TLS
.UseTls(tlsConfig)           // Custom TLS config
.UseMutualTls(caCert, clientCert, clientKey)  // mTLS

SASL Authentication

.WithSaslPlain("username", "password")
.WithSaslScramSha256("username", "password")
.WithSaslScramSha512("username", "password")
.WithGssapi(gssapiConfig)
.WithOAuthBearer(oauthConfig)
.WithOAuthBearerJwtBearer(options =>
{
    options.TokenEndpoint = "https://auth.example.com/oauth2/token";
    options.ClientId = "my-kafka-client";
    options.PrivateKey = rsaOrEcdsaPrivateKey;
    options.Audience = "kafka";
    options.Scopes = ["kafka:produce"];
})

Serialization

WithKeySerializer / WithValueSerializer

Custom serializers:

.WithKeySerializer(new JsonSerializer<OrderKey>())
.WithValueSerializer(new JsonSerializer<Order>())

Networking

WithConnectionsPerBroker

Number of TCP connections to each broker:

.WithConnectionsPerBroker(3)  // 3 parallel connections per broker

Default: 1. Must be 1 for idempotent producers (partition affinity requires a fixed connection).

WithConnectionsMaxIdle

Maximum time an unused broker connection stays open before the client closes it:

.WithConnectionsMaxIdle(TimeSpan.FromMinutes(9)) // Default: 540000ms
.WithConnectionsMaxIdle(Timeout.InfiniteTimeSpan) // Disable idle reaping

The default is 9 minutes, slightly below Kafka's broker-side connections.max.idle.ms default of 10 minutes. This lets Dekaf close unused connections first and avoid a request racing a broker idle close.

WithConnectionTimeout

Maximum time allowed for socket connection setup, including TLS and SASL handshakes:

.WithConnectionTimeout(TimeSpan.FromSeconds(10))

Set a larger maximum to enable KIP-601 adaptive setup deadlines. Consecutive failures grow the effective timeout exponentially with ±20% jitter, capped at the maximum. A successful setup resets it. Reconnect backoff remains a separate delay between attempts.

.WithConnectionTimeout(TimeSpan.FromSeconds(10))
.WithConnectionTimeoutMax(TimeSpan.FromSeconds(127))

When only WithConnectionTimeout is set, the maximum follows the initial value, preserving fixed-timeout behavior. Failure progression follows the broker ID plus advertised host:port. DNS address rotation for that broker and endpoint retains it; a different broker ID or advertised endpoint starts fresh.

WithTcpKeepAlive

Enable, disable, or tune TCP keepalive probes:

.WithTcpKeepAlive(false) // Disable keepalive
.WithTcpKeepAlive(TimeSpan.FromMinutes(2), TimeSpan.FromSeconds(30), retryCount: 3)

WithRemoteCertificateValidationCallback

Attach a custom TLS certificate validation callback for pinning or private PKI. Setting a callback enables TLS for the connection.

.WithRemoteCertificateValidationCallback((sender, cert, chain, errors) =>
    errors == SslPolicyErrors.None)

WithAdaptiveConnections

Configure adaptive connection scaling. When sustained buffer backpressure is detected, the producer automatically adds connections per broker to increase drain throughput:

// Use defaults (max 10 connections per broker)
.WithAdaptiveConnections()

// Custom maximum
.WithAdaptiveConnections(maxConnections: 5)

Adaptive scaling is enabled by default for non-idempotent producers. It monitors three signals before scaling up:

Pressure delta: at least 100 buffer-full events since the last check
Utilization: buffer is over 80% full
Cooldown: at least 30 seconds since the last scale-up

Connections are only scaled up, never down. Connections added during a traffic spike persist for the lifetime of the producer. Idempotent producers ignore this setting.

WithoutAdaptiveConnections

Disable adaptive scaling and use a fixed connection count:

.WithoutAdaptiveConnections()

WithBufferMemory

Maximum memory the producer uses for buffering unsent messages:

.WithBufferMemory(256 * 1024 * 1024)  // 256MB

Default: 2GB. When the buffer is full, ProduceAsync and Send block until space is freed (controlled by WithMaxBlockMs). Increase if profiling shows significant time in backpressure waits; decrease in memory-constrained environments.

WithBufferMemoryAllocationStrategy

Controls when each partition batch allocates its record storage:

.WithBufferMemoryAllocationStrategy(BufferMemoryAllocationStrategy.Incremental)

Full (the default) reserves one contiguous arena when a batch starts. Incremental rents pooled chunks as records arrive, which substantially reduces retained memory when a producer has many active partitions whose batches are only partly filled. Records remain zero-copy: compression reads the chunk sequence directly, and unencrypted single-batch sends use TCP scatter/gather. BufferMemory, batch limits, oversized-record behavior, retries, idempotence, and transactions have identical semantics under both strategies.

WithDeliveryLatencyTarget

Soft target for per-broker queueing latency (append to broker acknowledgement):

.WithDeliveryLatencyTarget(TimeSpan.FromMilliseconds(10)) // default
.WithDeliveryLatencyTarget(TimeSpan.Zero)                 // disable the bound

Default: 10ms. Before a broker's first successful acknowledgement, the producer admits one configured batch per current connection. This prevents an unsampled startup burst from filling the full pipeline. The adaptive controller then starts from a wider request window and probes up and down for the smallest whole-request window that preserves acknowledged goodput without increasing controllable queueing delay. Acks.None keeps the normal controller window because no acknowledgement can end the startup phase. When a broker reaches its budget, produce calls block exactly like BufferMemory exhaustion (subject to WithMaxBlock and cancellation). Raise the target if you prefer deeper buffering over latency; set TimeSpan.Zero to disable the bound.

WithSocketSendBufferBytes / WithSocketReceiveBufferBytes

TCP socket buffer sizes:

.WithSocketSendBufferBytes(1_048_576)    // 1MB send buffer
.WithSocketReceiveBufferBytes(1_048_576) // 1MB receive buffer

Observability

WithLoggerFactory

Enable logging:

.WithLoggerFactory(loggerFactory)

All Options Reference

Method	Default	Description
`WithBootstrapServers`	(required)	Broker addresses
`WithClientId`	"dekaf-producer"	Client identifier
`WithClientDnsLookup`	UseAllDnsIps	DNS lookup mode
`WithBootstrapResolveTimeout`	120000ms	Initial bootstrap DNS retry deadline
`WithAcks`	All	Acknowledgment mode
`WithLinger`	0ms	Batch wait time
`WithBatchSize`	1048576	Max batch size in bytes
`WithIdempotence`	true	Prevent duplicates
`WithTransactionalId`	null	Transaction ID
`WithTransactionTimeout`	60000ms	Transaction timeout
`UseCompression`	None	Compression codec
`WithCompressionLevel`	null	Codec-specific compression level
`WithPartitioner`	Default	Partition strategy
`WithAdaptivePartitioning`	true	Adapt sticky partition choices to queued broker load
`WithPartitionerAvailabilityTimeout`	0ms	Exclude backed-up partitions after timeout; 0 disables
`WithPartitionerIgnoreKeys`	false	Ignore keys for built-in sticky partitioning
`WithClientRack`	null	Producer rack for rack-aware partitioning
`WithRackAwarePartitioning`	false	Prefer partition leaders in the producer rack
`WithConnectionsPerBroker`	1	TCP connections per broker
`WithConnectionsMaxIdle`	540000ms	Close unused broker connections; `Timeout.InfiniteTimeSpan` disables
`WithConnectionTimeout`	30000ms	Socket connection setup timeout
`WithConnectionTimeoutMax`	Same as initial	Maximum adaptive connection setup timeout
`WithTcpKeepAlive`	enabled	TCP keepalive; 2m idle, 30s interval, 3 retries
`WithAdaptiveConnections`	enabled (max 10)	Auto-scale connections under load
`WithoutAdaptiveConnections`	-	Disable adaptive scaling
`WithBufferMemory`	auto-tuned	Max buffer for unsent messages
`WithBufferMemoryAllocationStrategy`	Full	Full arena or pooled incremental chunks
`WithDeliveryLatencyTarget`	10ms	Per-broker queueing latency target; `TimeSpan.Zero` disables
`WithMaxBlock`	60000ms	Max time produce calls wait for metadata or buffer space
`WithDeliveryTimeout`	120000ms	Max time for delivery success or failure
`WithRequestTimeout`	30000ms	Per-request timeout
`WithSocketSendBufferBytes`	OS default	TCP send buffer size
`WithSocketReceiveBufferBytes`	OS default	TCP receive buffer size
`UseTls`	false	Enable TLS
`WithRemoteCertificateValidationCallback`	null	Custom TLS certificate validation
`WithKeySerializer`	inferred	Key serializer
`WithValueSerializer`	inferred	Value serializer

Configuration Binding​

Connection Settings​

WithBootstrapServers​

WithClientId​

Delivery Settings​

WithAcks​

WithIdempotence​

Batching Settings​

WithLinger​

WithBatchSize​

Compression​

UseCompression / Specific Methods​

Partitioning​

WithPartitioner​

Transactions​

WithTransactionalId​

WithTwoPhaseCommit​

Security​

UseTls​

SASL Authentication​

Serialization​

WithKeySerializer / WithValueSerializer​

Networking​

WithConnectionsPerBroker​

WithConnectionsMaxIdle​

WithConnectionTimeout​

WithTcpKeepAlive​

WithRemoteCertificateValidationCallback​

WithAdaptiveConnections​

WithoutAdaptiveConnections​

WithBufferMemory​

WithBufferMemoryAllocationStrategy​

WithDeliveryLatencyTarget​

WithSocketSendBufferBytes / WithSocketReceiveBufferBytes​

Observability​

WithLoggerFactory​

All Options Reference​