Consumer Groups

Consumer groups enable multiple consumer instances to share the work of consuming a topic. Kafka automatically distributes partitions among group members.

Consumer groups guarantee per-partition ordering and give you full offset control (seek, replay, reset), but parallelism is capped at the partition count and one unprocessable message blocks the partition behind it. If you need work-queue semantics instead — more workers than partitions, per-record retry and redelivery, no ordering requirement — see Share Consumers (KIP-932), which includes a side-by-side comparison of the two models.

How Consumer Groups Work

When multiple consumers share a group ID:

1. Kafka assigns partitions to consumers (1 partition = 1 consumer max)
2. Each message is delivered to exactly one consumer in the group
3. If a consumer fails, its partitions are reassigned to others

Topic with 4 partitions:

Group "my-group" with 2 consumers:
  Consumer A: [Partition 0] [Partition 1]
  Consumer B: [Partition 2] [Partition 3]

Group "my-group" with 4 consumers:
  Consumer A: [Partition 0]
  Consumer B: [Partition 1]
  Consumer C: [Partition 2]
  Consumer D: [Partition 3]

Creating Consumer Group Members

Each consumer instance needs the same group ID:

using Dekaf;

// Instance 1
var consumer1 = await Kafka.CreateConsumer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("order-processors")  // Same group ID
    .SubscribeTo("orders")
    .BuildAsync();

// Instance 2 (different machine/process)
var consumer2 = await Kafka.CreateConsumer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("order-processors")  // Same group ID
    .SubscribeTo("orders")
    .BuildAsync();

Server-side Pattern Subscriptions

Kafka 4.1+ brokers can evaluate topic name patterns during group coordination:

var consumer = await Kafka.CreateConsumer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("order-processors")
    .SubscribeToPattern("orders-.*")
    .BuildAsync();

The pattern is sent through ConsumerGroupHeartbeat v1 as SubscribedTopicRegex. Kafka evaluates it with RE2/J-compatible syntax, and Dekaf does not translate .NET regex syntax.

For arbitrary .NET predicates, use consumer.Subscribe(Func<string, bool>) after building the consumer. That mode is client-side and polls metadata for matching topics, so it works with older brokers but does not use broker-side regex subscription.

Rebalancing

When the group membership changes, Kafka rebalances partitions:

• A new consumer joins
• A consumer leaves (graceful shutdown)
• A consumer is considered dead (heartbeat timeout)
• Topic partition count changes

Rebalance Listener

Get notified when partitions are assigned, revoked, lost, or stopped during graceful close:

using Dekaf;

var consumer = await Kafka.CreateConsumer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("my-group")
    .WithRebalanceListener(new MyRebalanceListener())
    .WithPartitionStopTimeout(TimeSpan.FromSeconds(30))
    .BuildAsync();

public sealed class MyRebalanceListener : IRebalanceListener, IPartitionStopListener
{
    public ValueTask OnPartitionsAssignedAsync(
        IEnumerable<TopicPartition> partitions,
        CancellationToken ct)
    {
        Console.WriteLine($"Assigned: {string.Join(", ", partitions)}");
        // Initialize resources for these partitions
        return ValueTask.CompletedTask;
    }

    public ValueTask OnPartitionsRevokedAsync(
        IEnumerable<TopicPartition> partitions,
        CancellationToken ct)
    {
        Console.WriteLine($"Revoked: {string.Join(", ", partitions)}");
        // Commit completed offsets, clean up resources
        return ValueTask.CompletedTask;
    }

    public ValueTask OnPartitionsLostAsync(
        IEnumerable<TopicPartition> partitions,
        CancellationToken ct)
    {
        Console.WriteLine($"Lost: {string.Join(", ", partitions)}");
        // Partitions were taken away. Do not commit offsets here.
        return ValueTask.CompletedTask;
    }

    public ValueTask OnPartitionsStoppedAsync(
        IEnumerable<TopicPartition> partitions,
        CancellationToken ct)
    {
        Console.WriteLine($"Stopped: {string.Join(", ", partitions)}");
        // Normal shutdown: drain and dispose partition-scoped resources
        return ValueTask.CompletedTask;
    }
}

Callback semantics:

Callback	When it runs	Offset rule
OnPartitionsAssignedAsync	After the group assigns partitions to this consumer.	Initialize partition-scoped state before records are processed.
OnPartitionsRevokedAsync	During cooperative rebalance before ownership is transferred.	Commit only offsets for records that have completed processing, then dispose partition-scoped state.
OnPartitionsLostAsync	After ownership was lost involuntarily, such as heartbeat timeout or unknown member recovery.	Do not commit offsets for lost partitions unless your application has a separate ownership guarantee.
OnPartitionsStoppedAsync	During graceful CloseAsync or DisposeAsync, after heartbeat, leader-refresh, auto-commit, and prefetch tasks stop and before final auto-commit, LeaveGroup, assignment cleanup, and resource disposal.	Drain local work if needed, commit completed offsets, then release resources.

Non-cancellation callback exceptions are logged and suppressed. The callback token is cancelled by caller cancellation, the aggregate WithDefaultApiTimeout, or the configured WithPartitionStopTimeout.

During a cooperative revoke, Dekaf awaits its automatic revoked-offset commit and OnPartitionsRevokedAsync before completing the assignment transfer. The initial KIP-848 heartbeat advertises ConsumerOptions.RebalanceTimeoutMs (60 seconds by default), which is the broker-visible window for completing that rebalance. The automatic revoked-offset commit is cancelled at the same limit. Listener callbacks receive the consumer operation/lifetime token rather than a separate timeout token, so revoke work should still complete within RebalanceTimeoutMs; exceeding the broker window can cause the member to lose its assignment.

MaxPollIntervalMs is separate: it limits time between foreground polls and is not sent as the rebalance timeout. Configuration-bound consumers can set the window with the RebalanceTimeoutMs key.

Use IConsumerAwareRebalanceListener when a callback must operate on the consumer without capturing its unrestricted instance:

public sealed class ConsumerAwareListener : IConsumerAwareRebalanceListener
{
    public ValueTask OnPartitionsAssignedAsync(
        IRebalanceConsumer consumer,
        IEnumerable<TopicPartition> partitions,
        CancellationToken ct)
    {
        var assigned = partitions.ToArray();
        consumer.SeekToBeginning(assigned);
        consumer.Pause(assigned);
        return ValueTask.CompletedTask;
    }

    public async ValueTask OnPartitionsRevokedAsync(
        IRebalanceConsumer consumer,
        IEnumerable<TopicPartition> partitions,
        CancellationToken ct)
    {
        var completed = GetCompletedOffsets(partitions);
        await consumer.CommitAsync(completed, ct);
    }

    public ValueTask OnPartitionsLostAsync(
        IRebalanceConsumer consumer,
        IEnumerable<TopicPartition> partitions,
        CancellationToken ct) => ValueTask.CompletedTask;
}

var consumer = await Kafka.CreateConsumer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("my-group")
    .WithRebalanceListener(new ConsumerAwareListener())
    .BuildAsync();

IRebalanceConsumer exposes commit/store, position and seek, pause/resume, assignment, group metadata, and offset queries. It intentionally excludes consume, close/dispose, subscribe, and assignment mutation. The view is valid only while its callback is running; using a retained view afterward throws InvalidOperationException.

Existing IRebalanceListener implementations and registrations require no changes. Migrate only listeners that need safe consumer operations by changing the implemented interface and adding the IRebalanceConsumer callback parameter.

For low-level consumers, track completed offsets only after durable processing. On revoke or graceful stop, commit those completed offsets. On lost, remove local state without committing:

using System.Collections.Concurrent;
using Dekaf;

public sealed class RebalanceCommitListener(
    ConcurrentDictionary<TopicPartition, TopicPartitionOffset> completedOffsets,
    Func<IReadOnlyCollection<TopicPartitionOffset>, CancellationToken, ValueTask> commitCompletedAsync)
    : IRebalanceListener, IPartitionStopListener
{
    public ValueTask OnPartitionsAssignedAsync(
        IEnumerable<TopicPartition> partitions,
        CancellationToken ct)
    {
        foreach (var partition in partitions)
            completedOffsets.TryRemove(partition, out _);

        return ValueTask.CompletedTask;
    }

    public async ValueTask OnPartitionsRevokedAsync(
        IEnumerable<TopicPartition> partitions,
        CancellationToken ct)
    {
        await CommitCompletedForAsync(partitions, ct);
    }

    public ValueTask OnPartitionsLostAsync(
        IEnumerable<TopicPartition> partitions,
        CancellationToken ct)
    {
        foreach (var partition in partitions)
            completedOffsets.TryRemove(partition, out _);

        return ValueTask.CompletedTask;
    }

    public async ValueTask OnPartitionsStoppedAsync(
        IEnumerable<TopicPartition> partitions,
        CancellationToken ct)
    {
        await CommitCompletedForAsync(partitions, ct);
    }

    private async ValueTask CommitCompletedForAsync(
        IEnumerable<TopicPartition> partitions,
        CancellationToken ct)
    {
        var offsets = new List<TopicPartitionOffset>();

        foreach (var partition in partitions)
        {
            if (completedOffsets.TryRemove(partition, out var offset))
                offsets.Add(offset);
        }

        if (offsets.Count > 0)
            await commitCompletedAsync(offsets, ct);
    }
}

Update the tracker from the consume loop after work succeeds:

completedOffsets[new TopicPartition(message.Topic, message.Partition)] =
    new TopicPartitionOffset(
        message.Topic,
        message.Partition,
        message.Offset + 1,
        message.LeaderEpoch ?? -1);

For the built-in partitioned runtime, prefer RunPartitionedAsync. It owns the channel-per-partition pattern, bounded queues, pause/resume backpressure, and revoke/lost/shutdown commits for you.

Rebalance Protocols and Assignors

Dekaf group subscriptions use Kafka's KIP-848 consumer group protocol, which requires Kafka 4.0 or later. KIP-848 assignment is server-side and every rebalance is incremental/cooperative: only partitions that move are revoked, while unaffected partitions continue processing.

Classic protocol support decision

Dekaf will not add Classic JoinGroup / SyncGroup membership. This is an intentional compatibility boundary, not a missing configuration switch:

• KIP-848 has been generally available since Kafka 4.0, which is already Dekaf's broker compatibility floor. It removes the group-wide synchronization barrier and moves assignment state to the broker.
• Apache Kafka's accepted KIP-1274 recommends the Consumer protocol in Kafka 4.3, changes the Java consumer default and deprecates Classic in Kafka 5.0, and removes public KafkaConsumer Classic support in Kafka 6.0. These are planned milestones under the accepted KIP-1274 roadmap, not fixed release commitments. Adding a second coordinator now would target an upstream client protocol planned for removal.
• Kafka supports online upgrade and downgrade between compatible Classic and Consumer group members. Dekaf tests both directions, so rolling migration does not require a permanent Classic implementation in Dekaf.
• One membership state machine avoids duplicating fencing, static-membership, commit, rebalance-listener, and shutdown semantics. This keeps the supported path easier to audit and optimize.

Applications that require Kafka 3.x, custom client-side assignor metadata, or a broker with Consumer groups disabled should use a Classic-compatible client while migrating. Manual assignment bypasses group membership, but it does not lower Dekaf's supported Kafka 4.0 broker floor. enforceRebalance is also intentionally absent and has no direct KIP-848 equivalent. Subscribe, SubscribePattern, and Unsubscribe update the subscription and trigger normal broker reconciliation. GroupRemoteAssignor is configured when creating the consumer; recreate the consumer to change it. Recreating a member changes group membership but does not guarantee an assignment change, so it is not a force-rebalance replacement.

Reconsider this decision only with explicit maintainer approval and concrete evidence that all of these are true:

1. A supported deployment cannot use Kafka 4.0+ Consumer groups or Kafka's online migration.
2. A broker-side assignor or temporary Classic-compatible client cannot cover the requirement.
3. The demand justifies a second coordinator implementation and its full integration matrix.
4. Apache Kafka's Classic removal roadmap has materially changed, or Dekaf intentionally adopts a different long-term compatibility policy.

See Apache Kafka's current consumer rebalance protocol guide for server settings, migration constraints, and configurations that no longer apply under KIP-848.

Choose one of the broker's configured remote assignors with WithGroupRemoteAssignor:

using Dekaf;

var consumer = await Kafka.CreateConsumer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("my-group")
    .WithGroupRemoteAssignor("uniform") // Or "range"
    .BuildAsync();

When no remote assignor is specified, the broker selects the first entry in its group.consumer.assignors configuration (uniform by default on Kafka 4.0).

Assignment algorithm and rebalance protocol are separate choices:

Group mode	Assignment	Rebalance semantics	Dekaf configuration
KIP-848 uniform	Broker distributes partitions as evenly as possible across subscribed members.	Incremental/cooperative	WithGroupRemoteAssignor("uniform")
KIP-848 range	Broker assigns ordered partition ranges per topic, preserving co-partitioning where possible.	Incremental/cooperative	WithGroupRemoteAssignor("range")
Classic range or roundrobin	A classic client-side leader computes the assignment.	Eager: every member revokes its full assignment.	Not supported by Dekaf; use a classic compatibility client.
Classic cooperative-sticky	A classic client-side leader computes a sticky assignment.	Cooperative, using the classic JoinGroup/SyncGroup APIs.	Not supported by Dekaf. It is not an alias for KIP-848 uniform.

Kafka 4.0 supports online migration between compatible Classic and Consumer groups. Dekaf's compatibility suite exercises both migration directions under the broker's default bidirectional migration policy. A classic range member can coexist temporarily with Dekaf while a rolling migration is in progress: the group becomes a Consumer group while a KIP-848 member is present and returns to Classic after the last KIP-848 member leaves. Custom classic assignor metadata or a restrictive group.consumer.migration.policy can prevent that migration. See Apache Kafka's consumer rebalance protocol guide for broker settings and limitations.

Static Membership

For faster rebalances with planned restarts, use static membership:

using Dekaf;

var consumer = await Kafka.CreateConsumer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("my-group")
    .WithGroupInstanceId("instance-1")  // Must be unique within the group
    .BuildAsync();

Benefits:

• Consumer can rejoin and get the same partitions back
• No rebalance if consumer restarts within session timeout
• Great for rolling deployments

warning

Each instance in the group must have a unique GroupInstanceId. Using the same ID causes fencing.

Session and Heartbeat Configuration

using Dekaf;

var consumer = await Kafka.CreateConsumer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("my-group")
    .WithSessionTimeout(TimeSpan.FromSeconds(45))    // Max time before considered dead
    .WithHeartbeatInterval(TimeSpan.FromSeconds(3)) // How often to send heartbeats
    .BuildAsync();

Guidelines:

• SessionTimeout should be > 3x HeartbeatInterval
• Longer timeout = more time for slow consumers, but slower failure detection
• Shorter timeout = faster failure detection, but more spurious rebalances

Consumer Group Metadata

Access information about the group:

// Get member ID
string? memberId = consumer.MemberId;

// Get consumer group metadata (for transactions)
var metadata = consumer.ConsumerGroupMetadata;

Scaling Consumers

Adding Consumers

When you add consumers to a group:

1. New consumer joins and triggers rebalance
2. Partitions are redistributed
3. New consumer starts receiving messages

Removing Consumers

When a consumer leaves gracefully (await using or CloseAsync):

1. Heartbeat, leader-refresh, auto-commit, and prefetch tasks stop
2. IPartitionStopListener.OnPartitionsStoppedAsync runs with the current assignment, if implemented by the configured rebalance listener
3. Final auto-commit runs when auto-commit mode has dirty offsets
4. Consumer sends LeaveGroup and releases resources
5. Remaining consumers get its partitions

The stop callback timeout defaults to five seconds and is configured with WithPartitionStopTimeout. On expiry, Dekaf cancels the callback token and stops awaiting the callback so assignment cleanup and resource disposal can continue. A callback that ignores cancellation may keep running after close, so it must not use consumer-owned resources after its token is cancelled. Caller cancellation and the aggregate WithDefaultApiTimeout can end the window sooner; those cancellations are re-thrown by CloseAsync after local cleanup. ConsumerCloseOptions controls only whether group membership is retained or left and does not replace the callback timeout.

For KafkaConsumerService, KafkaConsumerServiceOptions.ShutdownTimeout independently caps how long the hosted service awaits consumer disposal. Set it longer than WithPartitionStopTimeout plus remaining close work when the host must observe callback completion; the generic host's shutdown timeout can impose a further outer cap.

Cancel the token passed to ConsumeAsync before closing when you need to stop a pending fetch promptly during shutdown.

Maximum Parallelism

The maximum number of active consumers in a group equals the number of partitions:

Topic with 4 partitions:
  - 1 consumer: processes all 4 partitions
  - 2 consumers: each processes 2 partitions
  - 4 consumers: each processes 1 partition
  - 5 consumers: one consumer is idle!

Multiple Consumer Groups

Different groups consume the same topic independently:

using Dekaf;

// Analytics group - processes all messages
var analyticsConsumer = await Kafka.CreateConsumer<string, string>()
    .WithGroupId("analytics")
    .SubscribeTo("orders")
    .BuildAsync();

// Notification group - also processes all messages
var notificationConsumer = await Kafka.CreateConsumer<string, string>()
    .WithGroupId("notifications")
    .SubscribeTo("orders")
    .BuildAsync();

Each group:

• Tracks its own offsets
• Receives all messages
• Scales independently

Complete Example

using Dekaf;

public class OrderProcessor
{
    private readonly ILogger<OrderProcessor> _logger;

    public async Task RunAsync(string instanceId, CancellationToken ct)
    {
        await using var consumer = await Kafka.CreateConsumer<string, Order>()
            .WithBootstrapServers("localhost:9092")
            .WithGroupId("order-processors")
            .WithGroupInstanceId(instanceId)  // Static membership
            .WithRebalanceListener(new LoggingRebalanceListener(_logger))
            .WithOffsetCommitMode(OffsetCommitMode.Manual)
            .SubscribeTo("orders")
            .BuildAsync();

        _logger.LogInformation(
            "Consumer {InstanceId} started, member ID: {MemberId}",
            instanceId,
            consumer.MemberId
        );

        await foreach (var batch in consumer.ConsumeAsync(ct).Batch(100))
        {
            _logger.LogInformation(
                "Processing batch of {Count} orders from partitions: {Partitions}",
                batch.Count,
                string.Join(", ", batch.Select(m => m.Partition).Distinct())
            );

            foreach (var msg in batch)
            {
                await ProcessOrderAsync(msg.Value);
            }

            await consumer.CommitAsync();
        }
    }

    private class LoggingRebalanceListener : IRebalanceListener
    {
        private readonly ILogger _logger;

        public LoggingRebalanceListener(ILogger logger) => _logger = logger;

        public ValueTask OnPartitionsAssignedAsync(IEnumerable<TopicPartition> partitions, CancellationToken ct)
        {
            _logger.LogInformation("Partitions assigned: {Partitions}", string.Join(", ", partitions));
            return ValueTask.CompletedTask;
        }

        public ValueTask OnPartitionsRevokedAsync(IEnumerable<TopicPartition> partitions, CancellationToken ct)
        {
            _logger.LogInformation("Partitions revoked: {Partitions}", string.Join(", ", partitions));
            return ValueTask.CompletedTask;
        }

        public ValueTask OnPartitionsLostAsync(IEnumerable<TopicPartition> partitions, CancellationToken ct)
        {
            _logger.LogWarning("Partitions lost: {Partitions}", string.Join(", ", partitions));
            return ValueTask.CompletedTask;
        }
    }

    private Task ProcessOrderAsync(Order order) => Task.Delay(10);
}