Skip to main content

Share Consumers (KIP-932)

Share consumers implement KIP-932 "Queues for Kafka". Instead of assigning each partition to exactly one group member, a share group lets every member consume from any partition, with the broker handing out records under short-lived acquisition locks. Each record is acknowledged individually — accepted, released for redelivery, or rejected — giving you traditional message-queue semantics on top of Kafka topics.

Consumer or Share Consumer?​

The two models differ in who owns partitions and how progress is tracked:

Consumer groupShare group
Partition ownershipEach partition assigned to exactly one memberNone — any member fetches from any partition
Max parallelismPartition count (extra consumers idle)Unlimited — scale consumers past partition count
OrderingGuaranteed within a partitionNot guaranteed; records from one partition process concurrently
Progress trackingCommitted offset per partitionPer-record acknowledgement (Accept / Release / Reject)
Failure handlingCoarse: reprocess from committed offset; one poison message blocks the partition behind itPer-record: release or reject one record, the rest keep flowing
Position controlSeek, pause, offset reset, replay historyNone — the broker manages the delivery window
Delivery countingNot trackedDeliveryCount per record, enabling max-attempts logic
Broker requirementKafka 4.0+Kafka 4.2+ with group.share.enable=true

Pick a regular consumer group when you need per-partition ordering, offset-based replay, or stream-processing semantics — event sourcing, changelog consumption, windowed aggregation.

Pick a share consumer when you want work-queue semantics — more workers than partitions, per-message retry without blocking neighbors, or you are replacing a queue system (RabbitMQ, SQS, Azure Service Bus) with Kafka.

If you are unsure, start with a regular consumer group: it is the standard Kafka model, has no broker feature flag, and supports the full offset toolbox. Reach for share groups when partition-count ceilings or head-of-line blocking become the actual problem.

Requirements​

Share groups require Kafka 4.2+ with share groups enabled on the broker:

group.share.enable=true

Creating a Share Consumer​

Use the fluent builder:

using Dekaf;

await using var consumer = await Kafka.CreateShareConsumer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("order-workers")   // Share group ID (required)
    .SubscribeTo("orders")
    .BuildAsync();

Or from a root KafkaClient when multiple clients share connections:

await using var kafka = Kafka.Connect("localhost:9092");

await using var consumer = await kafka.CreateShareConsumer<string, string>("order-workers")
    .SubscribeTo("orders")
    .BuildAsync();

Share groups do not support manual partition assignment — Subscribe is the only way to receive records. The share group coordinator decides which partitions each member fetches from; the current set is exposed via consumer.Assignment.

Borrowed batch delivery​

PollBatchesAsync is an optional capability exposed by Dekaf's share consumer through IKafkaShareBatchConsumer<TKey, TValue>. The extension method on IKafkaShareConsumer<TKey, TValue> throws NotSupportedException for implementations without this capability.

using Dekaf.ShareConsumer;

await foreach (var batch in shareConsumer.PollBatchesAsync(cancellationToken))
{
    foreach (var record in batch)
    {
        Process(record.Value);
        batch.Acknowledge(record, AcknowledgeType.Accept);
    }
}

The outer iterator fetches batches asynchronously; the inner foreach uses a struct enumerator and value-type record views. Record views and first-use acknowledgement state live in pooled batch storage. Batch wrappers, network requests, and acknowledgement wire vectors have costs per batch. Deserializers that create strings or other objects still allocate for those results.

Choose PollAsync or PollBatchesAsync for a consumer instance. Switching between these polling APIs on the same instance throws InvalidOperationException. The class-based PollAsync API continues to return independent record objects with its existing retention behavior.

Ownership and partial enumeration​

A batch lease ends when you dispose it, advance or dispose the outer iterator, close the consumer, or unsubscribe. Record payload and offset properties reject access after the lease ends. KeyBytes, ValueBytes, header memory, and deserializer results that borrow their input must be read or copied before that boundary; an already copied memory slice cannot check whether the lease has ended.

record.Headers enumerates ShareBatchHeader values in wire order, preserving duplicate keys and null values. Header keys are UTF-8 bytes in KeyUtf8, so reading headers does not require allocating strings. Call Encoding.UTF8.GetString(header.KeyUtf8.Span) only when you need a string key. Deserializers that request string-based header context may incur header materialization costs.

Enumeration is forward-only. A new enumerator resumes at the next record, and only enumerated records count as delivered. Breaking the inner loop does not acknowledge the remaining records. Count reports available records; DeliveredCount reports how many this lease delivered.

Call batch.Acknowledge(record, type) before the lease ends. CommitAsync can send those stored dispositions afterwards. Disposing a batch releases local ownership without sending or accepting records. In implicit mode, the next poll or commit accepts delivered records without an explicit disposition. Closing releases implicit dispositions that were never submitted, preserving explicit Accept, Release, and Reject decisions and the outcome of a previously attempted request.

Renew requires explicit acknowledgement mode and broker support for ShareFetch/ShareAcknowledge v2. A successful Renew retains the acquired payload internally and replays it through a new batch lease with the original DeliveryCount. Release requests broker redelivery, which increments that count. Acknowledgement failures retain pending state for retry; a newer decision overrides an older failed request. All batch and consumer operations require the same external synchronization as PollAsync.

Consuming and Acknowledging​

PollAsync returns an IAsyncEnumerable of acquired records:

await foreach (var record in shareConsumer.PollAsync(cancellationToken))
{
    try
    {
        await ProcessAsync(record.Value);
        shareConsumer.Acknowledge(record, AcknowledgeType.Accept);
    }
    catch (TransientException)
    {
        // Redeliver to any group member (this one or another)
        shareConsumer.Acknowledge(record, AcknowledgeType.Release);
    }
    catch (PoisonMessageException)
    {
        // Permanently reject - never redelivered
        shareConsumer.Acknowledge(record, AcknowledgeType.Reject);
    }
}

The three acknowledgement types:

TypeEffect
AcceptRecord processed successfully; removed from the share partition
ReleaseRecord returned to the group for redelivery (increments its delivery count)
RejectRecord is unprocessable; permanently discarded, never redelivered

ShareConsumeResult<TKey, TValue> carries the usual Topic, Partition, Offset, Key, Value, Headers, and Timestamp, plus DeliveryCount — how many times the broker has delivered this record (first delivery = 1). Use it to dead-letter records that keep failing:

if (record.DeliveryCount >= 5)
{
    await deadLetterProducer.ProduceAsync("orders-dlq", record.Key, record.Value);
    shareConsumer.Acknowledge(record, AcknowledgeType.Reject);
    return;
}

Acknowledgement Modes​

The mode controls what happens to records you do not explicitly acknowledge. It maps to Kafka's share.acknowledgement.mode:

await using var consumer = await Kafka.CreateShareConsumer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("order-workers")
    .WithAcknowledgementMode(ShareAcknowledgementMode.Explicit)
    .BuildAsync(cancellationToken);

Implicit (default): records from the previous poll that were not passed to Acknowledge are automatically accepted when the next PollAsync iteration or CommitAsync sends acknowledgements. Call Acknowledge(record, Release) or Reject before the next poll if a record must not be auto-accepted.

Explicit: only records passed to Acknowledge are acknowledged. Unacknowledged records stay locked until their acquisition lock expires, then return to the group for redelivery.

Acknowledgements are batched and piggy-backed onto the next ShareFetch. To flush them immediately without fetching more records, call:

await consumer.CommitAsync(cancellationToken);

Observing Acknowledgement Outcomes​

Register an acknowledgement commit callback when application bookkeeping must observe the broker's final result:

await using var consumer = await Kafka.CreateShareConsumer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("order-workers")
    .WithAcknowledgementCommitCallback(results =>
    {
        foreach (var result in results)
        {
            if (result.Exception is null)
            {
                Console.WriteLine($"Acknowledged {result.TopicPartition}: " +
                    $"{result.Offsets.Length} record(s)");
            }
            else
            {
                Console.Error.WriteLine(
                    $"Acknowledgement failed for {result.TopicPartition}: {result.Exception.Message}");
            }
        }
    })
    .BuildAsync();

One ShareAcknowledgementCommitResult is reported per topic-partition. Offsets are ascending, Succeeded is true when Exception is null, and results are ordered by topic (ordinal) then partition.

The result span is valid only while the callback runs. Copy individual result values when they must be retained. Each result's Offsets is an allocation-free view that supports indexed access, foreach, and CopyTo.

The callback covers both acknowledgement transports:

  • inline acknowledgements piggy-backed by PollAsync;
  • standalone acknowledgements sent by CommitAsync or the close/dispose flush.

Dekaf invokes it once after broker retries finish and after successful acknowledgements are applied and failed acknowledgements are requeued. If cancellation ends a commit, failed partitions are requeued and reported before OperationCanceledException reaches the caller. A callback exception is logged and ignored; it never replaces the broker outcome or changes retry state.

The callback runs synchronously on the thread continuing the poll, commit, or close operation. Keep it short and non-blocking. Re-entering the same consumer from the callback is unsupported; record work for later processing instead.

Acquisition Locks and Renewal​

Records are delivered under a broker-side acquisition lock (default 30 seconds, broker config group.share.record.lock.duration.ms). If the lock expires before the record is acknowledged, the broker redelivers it to another member. The active timeout is exposed via consumer.AcquisitionLockTimeoutMs.

For work that outlives the lock, renew it:

shareConsumer.Acknowledge(record, AcknowledgeType.Renew);
await shareConsumer.CommitAsync(cancellationToken); // Sends the renewal
// ...continue long-running processing, then Accept/Release/Reject as normal

Renewal requires explicit acknowledgement mode and brokers supporting ShareFetch/ShareAcknowledge v2; older brokers throw BrokerVersionException.

Configuration​

Common builder options beyond the connection/TLS/SASL settings shared with other clients:

OptionDefaultDescription
WithGroupId— (required)Share group ID
WithAcknowledgementModeImplicitImplicit vs explicit acknowledgement (share.acknowledgement.mode)
WithAcknowledgementCommitCallback—Reports ordered per-partition broker outcomes after retries and internal bookkeeping
WithShareAcquireModeBatchOptimizedBatchOptimized acquires along producer batch boundaries; RecordLimit strictly caps at MaxPollRecords (share.acquire.mode)
WithMaxPollRecords500Maximum records per poll
WithFetchMinBytes / WithFetchMaxBytes1 / 50 MiBBroker fetch accumulation bounds
WithMaxPartitionFetchBytes1 MiBPer-partition fetch cap
WithFetchMaxWaitMs200Max broker wait for FetchMinBytes
WithSessionTimeoutMs45000Coordinator removes the member without a heartbeat within this window
WithHeartbeatIntervalMs3000Initial heartbeat interval (broker may adjust)

Built-in client telemetry​

Share consumers publish the KIP-932 client metrics under org.apache.kafka.consumer.share. through broker-side telemetry. The broker selects metric prefixes. Recording starts when a supported broker subscribes to those metrics; application-only subscriptions leave the built-in recorder inactive.

The built-in metrics cover poll intervals and idle ratio; coordinator heartbeat latency, activity and rebalances; fetch latency, throttling, request counts, serialized bytes and records; and acknowledgement sends and errors. Totals use monotonic OTLP sums with the broker's requested cumulative or delta temporality. Rates measure activity since the previous export of that rate, while averages and maxima retain their observation history. Record and byte averages divide by requests observed while record metrics were selected; fetch-only subscription periods do not dilute those averages. Age gauges report -1 before the first observed poll or heartbeat. Metric points have no partition or group attributes; the broker associates the push with its assigned client instance identity.

For streaming polling, a poll is one acquisition round. Time between rounds includes application processing; the idle ratio includes coordination, fetching and record preparation, excluding time spent in application code between yields. Record and byte totals count successfully deserialized acquired records, including the prepared portion of a partially consumed batch. A failed parsing window contributes no records or bytes; partial preparation results remain local until the partition parser succeeds. Bytes include the encoded record and its length prefix, excluding record-batch headers and unacquired offsets. Local renewal replay does not increment fetched totals. Acknowledgement totals count submitted records per request attempt, excluding gap placeholders; retry attempts and failed partitions contribute their own send/error counts.

Measurements aggregate per request, partition parsing window or record batch. The recorder keeps a reusable sample per broker and creates exported metric objects only at telemetry collection. Existing classic polling allocations still apply; enabling telemetry does not make the classic API allocation-free.

Application telemetry​

Share consumers can publish application counters and gauges through broker-side telemetry. Register metrics on the builder or on a running consumer:

using System.Threading;
using Dekaf.ShareConsumer;
using Dekaf.Telemetry;

long completed = 0;
await using var consumer = await Kafka.CreateShareConsumer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("jobs")
    .RegisterMetricForSubscription(new ApplicationTelemetryMetric(
        "com.example.jobs.completed", ApplicationTelemetryMetricKind.Counter,
        () => Interlocked.Read(ref completed)))
    .BuildAsync();

consumer.RegisterMetricForSubscription(new ApplicationTelemetryMetric(
    "com.example.jobs.queue.depth", ApplicationTelemetryMetricKind.Gauge,
    () => 42));
consumer.UnregisterMetricFromSubscription("com.example.jobs.queue.depth");

The broker's client-metrics configuration selects metric name prefixes, collection interval, compression, and counter temporality. Supply a cumulative monotonic value for a counter; Dekaf computes deltas when requested. Observation callbacks run on the telemetry background loop, so keep them fast, non-blocking, and safe to call alongside application work. Metrics outside requested prefixes are not observed.

Registering the same name replaces its previous metric and resets counter history. Removing a missing name does nothing. Builders snapshot registrations for each built consumer; ShareConsumerOptions.ApplicationMetrics also supplies initial registrations. Metric attributes are copied when the metric is created. Registration and removal after consumer disposal throw ObjectDisposedException.

Runtime methods use the optional IApplicationTelemetryShareConsumer capability. Existing implementations of IKafkaShareConsumer<TKey, TValue> remain compatible; the extension methods throw NotSupportedException when that capability is absent. A supported broker receives the encoded application metrics under the client's assigned instance identity, including the final telemetry push during shutdown.

Thread Safety​

IKafkaShareConsumer<TKey, TValue> is not thread-safe. Call Subscribe, PollAsync, Acknowledge, CommitAsync, and Unsubscribe from a single thread or with external synchronization. Run multiple consumer instances for parallelism — that is the point of share groups.

Shutdown​

CloseAsync and DisposeAsync release delivered records that have not yet been implicitly acknowledged by the next poll or CommitAsync. This includes disposal when application processing throws and records left after partial enumeration. Records fetched or parsed but never yielded are not implicitly accepted; session closure releases their acquisition locks.

Explicitly selected Accept, Release and Reject outcomes are preserved. Outcomes already submitted by a previous poll/commit remain selected even if their failed request is retried during close. A pending Renew is attempted as a renewal, never as acceptance; session closure then releases remaining locks and stops local replay. Shutdown is best-effort: if cancellation or broker failure prevents release, records remain available for redelivery after the broker's acquisition lock expires.

Unsubscribe releases pending records and clears the subscription. To close:

await consumer.CloseAsync();
// or rely on await using for disposal

Administration​

IAdminClient covers share group operations: ListShareGroupsAsync, DescribeShareGroupsAsync, DeleteShareGroupsAsync, DescribeShareGroupOffsetsAsync, AlterShareGroupOffsetsAsync, and DeleteShareGroupOffsetsAsync.

Group deletion returns one result per requested ID, so a batch preserves partial failures instead of throwing away successful results:

var results = await admin.DeleteShareGroupsAsync(["jobs-a", "jobs-b"]);
foreach (var (groupId, result) in results)
{
    Console.WriteLine($"{groupId}: {result.ErrorCode}");
}

The operation uses the group coordinator and Kafka's DeleteGroups API, matching Kafka 4.3's deleteShareGroups implementation. Active groups normally return NonEmptyGroup; close their consumers before deletion.

Per-group results cover terminal error codes only. If a request keeps failing with a retriable error, the call throws after retries are exhausted and returns no results. Duplicate group IDs raise ArgumentException before any request is sent. Dekaf's built-in and in-memory admin clients expose deletion through IShareGroupDeletionAdminClient; the IAdminClient extension preserves the same call syntax for binary compatibility.

Migration note: implicit acknowledgement on shutdown​

Earlier versions treated outstanding implicit deliveries as Accept during close, which could acknowledge records whose application processing failed. Close and await-using disposal now release these deliveries. After successfully processing the final records, call CommitAsync before closing when they should be accepted, or explicitly call Acknowledge(record, AcknowledgeType.Accept) for each completed record. Do not commit from an unconditional finally block after failed processing. The in-memory share consumer follows the same provisional-delivery shutdown rule.

Testing​

Dekaf.Testing provides InMemoryShareConsumer<TKey, TValue> for broker-free unit tests, and AddDekafInMemory() swaps DI registrations for in-memory doubles. See Testing.