Consumer Options

Complete reference for all consumer configuration options.

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

// Before: DI examples only showed connection and group settings.
builder.Services.AddDekaf(dekaf =>
{
    dekaf.AddConsumer<string, string>(consumer => consumer
        .WithBootstrapServers("localhost:9092")
        .WithGroupId("orders"));
});

// After: DI uses the full consumer builder.
builder.Services.AddDekaf(dekaf =>
{
    dekaf.AddConsumer<string, string>(consumer => consumer
        .WithBootstrapServers("localhost:9092")
        .WithGroupId("orders")
        .WithFetchMinBytes(1024)
        .WithFetchMaxBytes(50 * 1024 * 1024)
        .WithPrefetchPipelineDepth(4)
        .WithSaslScramSha512("user", "password")
        .SubscribeTo("orders"));
});

Configuration Binding

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

{
  "Kafka": {
    "Consumers": {
      "Orders": {
        "BootstrapServers": [
          "broker1:9092",
          "broker2:9092"
        ],
        "ClientId": "orders-consumer",
        "GroupId": "orders",
        "AutoOffsetReset": "Earliest",
        "OffsetCommitMode": "Manual",
        "FetchMinBytes": 1024,
        "FetchMaxBytes": 52428800,
        "FetchBufferMemoryBytes": 104857600,
        "FetchMaxWaitMs": 200,
        "MaxPollRecords": 500,
        "UseTls": true
      }
    }
  }
}

builder.Services.AddDekaf(dekaf =>
{
    dekaf.AddConsumer<string, Order>(
        builder.Configuration.GetSection("Kafka:Consumers:Orders"),
        consumer => consumer
            .WithValueDeserializer(new JsonDeserializer<Order>())
            .SubscribeTo("orders"));
});

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
`WithGroupId(...)`	`GroupId`	String
`WithGroupInstanceId(...)`	`GroupInstanceId`	String
`WithGroupRemoteAssignor(...)`	`GroupRemoteAssignor`	Common values: `uniform`, `range`
`WithOffsetCommitMode(...)`	`OffsetCommitMode`	`Auto` or `Manual`
`WithOffsetStoreTiming(...)`	`OffsetStoreTiming`	`AfterProcessing` (default, at-least-once) or `OnDelivery` (at-most-once)
`WithAutoOffsetStore(...)`	`EnableAutoOffsetStore`	Boolean; disable for explicit `StoreOffset` acknowledgment
`WithAutoCommitInterval(...)`	`AutoCommitIntervalMs`	Milliseconds
`WithAutoOffsetReset(...)`	`AutoOffsetReset`	`Latest`, `Earliest`, `None`
`WithAutoOffsetResetByDuration(...)`	`AutoOffsetReset`, `AutoOffsetResetDuration`	Use `AutoOffsetReset: ByDuration` plus a duration, or Kafka-style `by_duration:PT24H`
`WithAutoOffsetResetNewPartitions(...)`	`AutoOffsetResetNewPartitions`	Optional KIP-1327 policy for newly-expanded partitions; `Latest` or `Earliest`
`WithAutoOffsetResetNewPartitionsByDuration(...)`	`AutoOffsetResetNewPartitions`, `AutoOffsetResetNewPartitionsDuration`	Optional independent duration policy for newly-expanded partitions
`WithFetchMinBytes(...)`	`FetchMinBytes`	Bytes
`WithFetchMaxBytes(...)`	`FetchMaxBytes`	Bytes
`WithFetchBufferMemory(...)`	`FetchBufferMemoryBytes`	Aggregate raw Fetch response bytes; default 100 MiB
`WithMaxPartitionFetchBytes(...)`	`MaxPartitionFetchBytes`	Bytes
`WithFetchMaxWait(...)`	`FetchMaxWaitMs`	Milliseconds
`WithFetchSessions(...)`	`EnableFetchSessions`	Boolean
`WithMaxPollRecords(...)`	`MaxPollRecords`	Integer
`WithSessionTimeout(...)`	`SessionTimeoutMs`	Milliseconds
`WithHeartbeatInterval(...)`	`HeartbeatIntervalMs`	Milliseconds
`WithIsolationLevel(...)`	`IsolationLevel`	`ReadUncommitted` or `ReadCommitted`
`WithPartitionEof(...)`	`EnablePartitionEof`	Boolean
`WithQueuedMinMessages(...)`	`QueuedMinMessages`	Integer
`WithQueuedMaxMessagesKbytes(...)`	`QueuedMaxMessagesKbytes`	KiB; omit to keep auto-tuning
`WithPrefetchPipelineDepth(...)`	`PrefetchPipelineDepth`	Integer
`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
`WithAdaptiveFetchSizing(...)`	`EnableAdaptiveFetchSizing`, `AdaptiveFetchSizingOptions`	Bind nested adaptive sizing fields
`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
`WithMetadataRecoveryStrategy(...)`	`MetadataRecoveryStrategy`	`None` or `Rebootstrap`
`WithMetadataClusterCheck(...)`	`MetadataClusterCheckEnabled`	KIP-1242 identity check; default `true`, ignored with `None` recovery
`WithMetadataRecoveryRebootstrapTrigger(...)`	`MetadataRecoveryRebootstrapTriggerMs`	Milliseconds

Topics, deserializers, rebalance listeners, interceptors, and retry policies are objects or runtime choices, so configure those in the fluent callback.

Connection Settings

WithBootstrapServers

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

.WithBootstrapServers("localhost:9092")
.WithBootstrapServers("broker1:9092", "broker2:9092")
.WithBootstrapServers("broker1:9092,broker2:9092")

WithClientId

Client identifier:

.WithClientId("order-processor")

Consumer Group Settings

WithGroupId

Consumer group identifier (required for group consumption):

.WithGroupId("order-processors")

WithGroupInstanceId

Static membership ID for faster rebalances:

.WithGroupInstanceId("instance-1")

Offset Management

WithOffsetCommitMode

How offsets are committed (matches Kafka's enable.auto.commit):

.WithOffsetCommitMode(OffsetCommitMode.Auto)    // Automatic commit in background (default)
.WithOffsetCommitMode(OffsetCommitMode.Manual)  // You call CommitAsync() explicitly

Processing Guarantees

Intent-level methods that configure commit mode, offset storage, and staging timing together (see Delivery Semantics):

.WithAtLeastOnceProcessing()  // The default, stated explicitly: offsets become committable
                              // only after the loop demonstrably processed the record
.WithAtMostOnceProcessing()   // Confluent-style: offsets committable at delivery,
                              // before processing runs
.WithAutoOffsetStore(false)   // Strict at-least-once: only offsets you pass to
                              // StoreOffset(...) are ever committed
.WithOffsetStoreTiming(OffsetStoreTiming.AfterProcessing) // Granular knob behind the
                              // intent methods (AfterProcessing | OnDelivery)

WithAutoCommitInterval

Control how often offsets are committed in Auto mode:

.WithAutoCommitInterval(TimeSpan.FromSeconds(5))  // Same, using TimeSpan

WithAutoOffsetReset

Where to start when no committed offset exists:

.WithAutoOffsetReset(AutoOffsetReset.Latest)    // New messages only (default)
.WithAutoOffsetReset(AutoOffsetReset.Earliest)  // From beginning
.WithAutoOffsetReset(AutoOffsetReset.None)      // Throw exception
.WithAutoOffsetResetByDuration(TimeSpan.FromHours(24))

Configuration can use either a separate duration value:

{
  "AutoOffsetReset": "ByDuration",
  "AutoOffsetResetDuration": "24:00:00"
}

or Kafka's ISO-8601 form:

{
  "AutoOffsetReset": "by_duration:PT24H"
}

For KIP-1327 brokers, newly-expanded partitions can use an independent policy:

.WithAutoOffsetResetNewPartitions(AutoOffsetReset.Earliest)
.WithAutoOffsetResetNewPartitionsByDuration(TimeSpan.FromHours(1))

Use AutoOffsetResetNewPartitions and AutoOffsetResetNewPartitionsDuration for configuration binding. If unset, newly-expanded partitions use the base policy above. Committed offsets still take precedence.

Fetch Settings

WithMaxPollRecords

Maximum messages per poll:

.WithMaxPollRecords(500)  // Default: 500

Fetch Tuning

Control how data is fetched from brokers:

.WithFetchMinBytes(1024)
.WithFetchMaxBytes(50 * 1024 * 1024)
.WithFetchBufferMemory(100L * 1024 * 1024)
.WithMaxPartitionFetchBytes(4 * 1024 * 1024)
.WithFetchMaxWait(TimeSpan.FromMilliseconds(200))
.WithFetchSessions(enabled: true)

WithFetchBufferMemory bounds the exact aggregate bytes held by raw Fetch responses across brokers and pipelined requests. It must be at least the configured FetchMaxBytes. An adaptive or oversized first-batch response larger than the limit is admitted only when it is the sole reservation. Coordinator responses do not consume this budget. Observe current pressure through dekaf.consumer.fetch_buffer.used_bytes, free_bytes, depleted_percent, and depleted_duration.

Session Settings

WithSessionTimeout

How long before consumer is considered dead:

.WithSessionTimeout(45000)                      // 45 seconds (default)
.WithSessionTimeout(TimeSpan.FromSeconds(45))  // Same, using TimeSpan

Subscription

SubscribeTo

Subscribe to topics during build:

.SubscribeTo("orders")
.SubscribeTo("orders", "payments", "notifications")

SubscribeToPattern

Subscribe with a broker-side topic name pattern:

.SubscribeToPattern("orders-.*")

Kafka evaluates the pattern on the broker using RE2/J-compatible syntax. Dekaf sends the pattern as-is; .NET regular expression syntax is not translated.

Server-side pattern subscription requires Kafka 4.1+ brokers with ConsumerGroupHeartbeat v1. Use Subscribe(Func<string, bool>) on IKafkaConsumer when you need arbitrary .NET predicates or compatibility with older brokers. That overload remains client-side and refreshes metadata to find matching topics.

Rebalancing

WithRebalanceListener

Get notified of partition changes. If the listener also implements IPartitionStopListener, OnPartitionsStoppedAsync runs during graceful CloseAsync or DisposeAsync with the current assignment before final auto-commit, LeaveGroup, assignment cleanup, and resource disposal:

.WithRebalanceListener(new MyRebalanceListener())

The same method accepts IConsumerAwareRebalanceListener. Its callbacks receive an IRebalanceConsumer view with safe commit, seek, pause/resume, metadata, and offset operations. The view is invalidated as soon as the callback completes.

The stop callback is best-effort and bounded to five seconds. If it observes shutdown cancellation, Dekaf still completes local cleanup before rethrowing the cancellation from CloseAsync.

Networking

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. Connections with in-flight requests are not reaped.

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 separate.

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

Security

UseTls

Enable TLS:

.UseTls()
.UseTls(tlsConfig)
.UseMutualTls(caCert, clientCert, clientKey)

Custom TLS certificate validation can be attached for pinning or private PKI. Setting a callback enables TLS for the connection.

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

SASL Authentication

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

Serialization

WithKeyDeserializer / WithValueDeserializer

Custom deserializers:

.WithKeyDeserializer(new JsonDeserializer<OrderKey>())
.WithValueDeserializer(new JsonDeserializer<Order>())

Advanced Settings

WithPartitionEof

Receive notification when reaching end of partition:

.WithPartitionEof(true)

WithIsolationLevel

For transactional reads:

.WithIsolationLevel(IsolationLevel.ReadCommitted)    // Only committed messages
.WithIsolationLevel(IsolationLevel.ReadUncommitted)  // All messages (default)

Observability

WithLoggerFactory

.WithLoggerFactory(loggerFactory)

All Options Reference

Method	Default	Description
`WithBootstrapServers`	(required)	Broker addresses
`WithClientId`	"dekaf-consumer"	Client identifier
`WithClientDnsLookup`	UseAllDnsIps	DNS lookup mode
`WithBootstrapResolveTimeout`	120000ms	Initial bootstrap DNS retry deadline
`WithGroupId`	null	Consumer group ID
`WithGroupInstanceId`	null	Static membership ID
`WithOffsetCommitMode`	Auto	Offset management mode
`WithOffsetStoreTiming`	AfterProcessing	When offsets become committable (at-least-once default)
`WithAutoOffsetStore`	true	Automatic offset staging
`WithAutoCommitInterval`	5000ms	Auto-commit interval
`WithAutoOffsetReset`, `WithAutoOffsetResetByDuration`	Latest	Start position
`WithFetchMinBytes`	1	Minimum fetch bytes
`WithFetchMaxBytes`	52428800	Maximum total fetch bytes
`WithFetchBufferMemory`	104857600	Aggregate queued and in-flight raw Fetch response limit
`WithMaxPartitionFetchBytes`	1048576	Maximum fetch bytes per partition
`WithFetchMaxWait`	200ms	Maximum fetch wait
`WithFetchSessions`	true	Enable incremental fetch sessions
`WithMaxPollRecords`	500	Max messages per poll
`WithSessionTimeout`	45000ms	Session timeout
`WithHeartbeatInterval`	3000ms	Group heartbeat interval
`SubscribeTo`	(none)	Topics to subscribe
`SubscribeToPattern`	(none)	Broker-side topic regex subscription; Kafka 4.1+
`WithRebalanceListener`	null	Rebalance callbacks
`WithPartitionEof`	false	EOF notifications
`WithQueuedMinMessages`	100000	Prefetch target count
`WithQueuedMaxMessagesKbytes`	auto-tuned	Prefetch memory limit
`WithPrefetchPipelineDepth`	3	Overlapping prefetch operations
`WithConnectionsMaxIdle`	540000ms	Close unused broker connections; `Timeout.InfiniteTimeSpan` disables
`WithConnectionsPerBroker`	2	TCP connections per broker
`WithAdaptiveConnections`	enabled (max 4)	Auto-scale connections under load
`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
`UseTls`	false	Enable TLS
`WithRemoteCertificateValidationCallback`	null	Custom TLS certificate validation

Configuration Binding​

Connection Settings​

WithBootstrapServers​

WithClientId​

Consumer Group Settings​

WithGroupId​

WithGroupInstanceId​

Offset Management​

WithOffsetCommitMode​

Processing Guarantees​

WithAutoCommitInterval​

WithAutoOffsetReset​

Fetch Settings​

WithMaxPollRecords​

Fetch Tuning​

Session Settings​

WithSessionTimeout​

Subscription​

SubscribeTo​

SubscribeToPattern​

Rebalancing​

WithRebalanceListener​

Networking​

WithConnectionsMaxIdle​

WithConnectionTimeout​

WithTcpKeepAlive​

Security​

UseTls​

SASL Authentication​

Serialization​

WithKeyDeserializer / WithValueDeserializer​

Advanced Settings​

WithPartitionEof​

WithIsolationLevel​

Observability​

WithLoggerFactory​

All Options Reference​