Producer Options

Complete reference for all producer configuration options.

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

using Dekaf.Compression;
using Dekaf.Compression.Lz4;

// Install Dekaf.Compression.Lz4 and register it before configuring clients.
CompressionCodecRegistry.Default.AddLz4();

// 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)

Install the matching package and explicitly register optional codecs at startup with CompressionCodecRegistry.Default.AddLz4(), AddZstd(), or AddSnappy() before building the producer. Import the matching namespace, such as Dekaf.Compression.Lz4. The builder methods above select the compression type. See Compression for complete registration examples.

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: auto-tuned from the process memory budget (see DekafMemoryBudget). When the buffer is full, ProduceAsync and FireAsync wait for space for up to the WithMaxBlock duration. 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.

With Full, every batch arena is BatchSize plus 12.5% (about 1.1MB with the default 1MB BatchSize), even when the batch holds one small record. Arenas are pinned and allocated without zeroing, so container RSS metrics undercount them while the GC heap limit counts them in full. Idle arenas are kept for reuse in a pool shared by every producer in the process. Each producer allows the smaller of its BufferMemory and 10% of the memory available to the GC (75% of the container memory limit by default), and the pool keeps no more than the largest allowance among live producers. In small containers with low-volume topics, a smaller WithBatchSize or the Incremental strategy keeps live batch memory proportional to the records buffered.

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. The target is end to end, so the configured WithLinger delay counts against it: the controller governs the remainder (target minus linger, never below half the target) as controllable queueing latency. With the default 10ms target and a 5ms linger, standing per-broker queueing is steered toward 5ms. 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 delivery latency target (append to ack, linger counts against it); 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