# Durable Consumer Groups (/learn/events-store/tutorials/consumer-groups)



Consumer groups in Events Store distribute event processing across multiple subscribers while maintaining durable position tracking. Each event is delivered to exactly one member of the group, and the group's position is preserved across reconnections.

<Callout type="info">
  For the underlying concept — why a consumer group gives you load-balancing and broadcast from the same channel — see [Scaling & flow](/learn/concepts/scaling-and-flow#consumer-groups-getting-both-from-one-channel) in the Fundamentals track. This page focuses on the Events Store specifics: durable position tracking and resume after disconnect.
</Callout>

## How Consumer Groups Work [#how-consumer-groups-work]

<Mermaid
  chart="graph LR
  P[&#x22;Publisher&#x22;]
  ST[(&#x22;Event Store<br/>channel: orders.processing&#x22;)]
  W1[&#x22;Worker A<br/>group: processors&#x22;]
  W2[&#x22;Worker B<br/>group: processors&#x22;]
  W3[&#x22;Worker C<br/>group: processors&#x22;]
  M[&#x22;Auditor<br/>no group&#x22;]

  P -- persist --> ST
  ST -- &#x22;round-robin&#x22; --> W1
  ST -- &#x22;round-robin&#x22; --> W2
  ST -- &#x22;round-robin&#x22; --> W3
  ST -- &#x22;all events&#x22; --> M

  class ST store
  class W1,W2,W3 queue
  class P,M client"
/>

*Each stored event is delivered to exactly one member of a group, while ungrouped subscribers receive every event independently.*

* **Group members** share the event stream: each event goes to exactly one member
* **Ungrouped subscribers** receive every event independently
* **Position is durable**: if all members disconnect, the group resumes from the last position when any member reconnects
* The durable name is `{channel}-{group}`

## Prerequisites [#prerequisites]

* KubeMQ server running on `localhost:50000`
* SDK installed ([Getting Started](/learn/events-store/getting-started))

## Step-by-Step [#step-by-step]

<Steps>
  <Step>
    ### Create the Consumer Group Workers [#create-the-consumer-group-workers]

    Specify the `group` parameter when subscribing. Subscribers with the same group name on the same channel form a consumer group.

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="order_worker.go"
        package main

        import (
            "context"
            "fmt"
            "log"
            "os"

            "github.com/kubemq-io/kubemq-go/v2"
        )

        func main() {
            workerID := os.Getenv("WORKER_ID")
            if workerID == "" {
                workerID = "worker-1"
            }

            ctx := context.Background()
            client, err := kubemq.NewClient(ctx,
                kubemq.WithAddress("localhost", 50000),
            )
            if err != nil {
                log.Fatal(err)
            }
            defer client.Close()

            sub, err := client.SubscribeToEventsStore(ctx,
                "orders.processing",
                "order-processors",
                kubemq.StartFromFirst(),
                kubemq.WithOnEvent(func(event *kubemq.Event) {
                    fmt.Printf("[%s] Processing seq=%d: %s\n",
                        workerID, event.Sequence, string(event.Body))
                }),
                kubemq.WithOnError(func(err error) {
                    log.Printf("[%s] Error: %v", workerID, err)
                }),
            )
            if err != nil {
                log.Fatal(err)
            }
            defer sub.Unsubscribe()

            log.Printf("[%s] Ready in group 'order-processors'", workerID)
            <-ctx.Done()
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="order_worker.py"
        import os
        import time
        from kubemq import (
            PubSubClient, EventsStoreSubscription,
            EventStoreStartPosition, CancellationToken,
        )

        worker_id = os.environ.get("WORKER_ID", "worker-1")

        def on_event(event):
            print(f"[{worker_id}] Processing seq={event.sequence}: "
                  f"{event.body.decode('utf-8')}")

        with PubSubClient(address="localhost:50000") as client:
            client.subscribe_to_events_store(
                subscription=EventsStoreSubscription(
                    channel="orders.processing",
                    group="order-processors",
                    start_position=EventStoreStartPosition.StartFromFirst,
                    on_receive_event_callback=on_event,
                    on_error_callback=lambda e: print(f"[{worker_id}] Error: {e}"),
                ),
                cancel=CancellationToken(),
            )
            print(f"[{worker_id}] Ready in group 'order-processors'")
            time.sleep(300)
        ```
      </Tab>

      <Tab value="Node.js">
        ```typescript title="order_worker.ts"
        import { KubeMQClient, EventStoreStartPosition } from 'kubemq-js';

        const workerId = process.env.WORKER_ID ?? 'worker-1';
        const client = await KubeMQClient.create({ address: 'localhost:50000' });

        client.subscribeToEventsStore({
          channel: 'orders.processing',
          group: 'order-processors',
          startFrom: EventStoreStartPosition.StartFromFirst,
          onEvent: (msg) =>
            console.log(
              `[${workerId}] Processing seq=${msg.sequence}: ` +
                `${new TextDecoder().decode(msg.body)}`
            ),
          onError: (err) =>
            console.error(`[${workerId}] Error:`, err.message),
        });

        console.log(`[${workerId}] Ready in group 'order-processors'`);
        ```
      </Tab>

      <Tab value="Java">
        ```java title="OrderWorker.java"
        String workerId = System.getenv().getOrDefault("WORKER_ID", "worker-1");

        PubSubClient client = PubSubClient.builder()
            .address("localhost:50000")
            .clientId(workerId)
            .build();

        client.subscribeToEventsStore(EventsStoreSubscription.builder()
            .channel("orders.processing")
            .group("order-processors")
            .startPosition(EventStoreStartPosition.StartFromFirst)
            .onReceiveEventCallback(event ->
                System.out.printf("[%s] Processing seq=%d: %s%n",
                    workerId, event.getSequence(), new String(event.getBody())))
            .onErrorCallback(err ->
                System.err.printf("[%s] Error: %s%n", workerId, err.getMessage()))
            .build());

        System.out.printf("[%s] Ready in group 'order-processors'%n", workerId);
        Thread.sleep(300_000);
        client.close();
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="OrderWorker.cs"
        var workerId = Environment.GetEnvironmentVariable("WORKER_ID") ?? "worker-1";

        await using var client = new KubeMQClient(new KubeMQClientOptions());
        await client.ConnectAsync();

        Console.WriteLine($"[{workerId}] Ready in group 'order-processors'");
        await foreach (var msg in client.SubscribeToEventsStoreAsync(
            new EventsStoreSubscription
            {
                Channel = "orders.processing",
                Group = "order-processors",
                StartPosition = EventStoreStartPosition.StartFromFirst,
            }))
        {
            Console.WriteLine($"[{workerId}] Processing seq={msg.Sequence}: "
                + $"{Encoding.UTF8.GetString(msg.Body.Span)}");
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="OrderWorker.kt"
        val workerId = System.getenv("WORKER_ID") ?: "worker-1"

        val client = KubeMQClient.pubSub {
            address = "localhost:50000"
            clientId = workerId
        }

        client.use {
            println("[$workerId] Ready in group 'order-processors'")
            client.subscribeToEventsStore {
                channel = "orders.processing"
                group = "order-processors"
                startPosition = StartPosition.StartFromFirst
            }.collect { msg ->
                println("[$workerId] Processing seq=${msg.sequence}: ${String(msg.body)}")
            }
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="order_worker.cc"
        const char* env_id = std::getenv("WORKER_ID");
        std::string worker_id = env_id ? env_id : "worker-1";

        kubemq::ClientOptions options;
        options.set_address("localhost", 50000);
        options.set_client_id(worker_id);
        auto client = kubemq::Client::Create(options).value();

        std::cout << "[" << worker_id << "] Ready in group 'order-processors'" << std::endl;
        client->SubscribeToEventsStore("orders.processing", "order-processors",
            kubemq::StartPosition::StartFromFirst,
            [&worker_id](const kubemq::EventStoreReceived& msg) {
                std::cout << "[" << worker_id << "] Processing seq=" << msg.sequence()
                          << ": " << msg.body() << std::endl;
            },
            [&worker_id](const std::string& err) {
                std::cerr << "[" << worker_id << "] Error: " << err << std::endl;
            });
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="order_worker.rs"
        use kubemq::prelude::*;
        use kubemq::EventsStoreSubscription;
        use std::time::Duration;

        #[tokio::main]
        async fn main() -> kubemq::Result<()> {
            let worker_id = std::env::var("WORKER_ID").unwrap_or_else(|_| "worker-1".to_string());

            let client = KubemqClient::builder()
                .host("localhost")
                .port(50000)
                .build()
                .await?;

            // Subscribers with the same group form a consumer group.
            let sub = client
                .subscribe_to_events_store(
                    "orders.processing",
                    "order-processors",
                    EventsStoreSubscription::StartFromFirst,
                    {
                        let worker_id = worker_id.clone();
                        move |event| {
                            let worker_id = worker_id.clone();
                            Box::pin(async move {
                                println!(
                                    "[{}] Processing seq={}: {}",
                                    worker_id,
                                    event.sequence,
                                    String::from_utf8_lossy(&event.body)
                                );
                            })
                        }
                    },
                    None,
                )
                .await?;

            println!("[{}] Ready in group 'order-processors'", worker_id);
            tokio::time::sleep(Duration::from_secs(300)).await;

            sub.unsubscribe().await;
            client.close().await?;
            Ok(())
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="order_worker.rb"
        require 'kubemq'

        worker_id = ENV.fetch('WORKER_ID', 'worker-1')

        client = KubeMQ::PubSubClient.new(address: 'localhost:50000', client_id: worker_id)
        cancel = KubeMQ::CancellationToken.new

        # Subscribers sharing the same group name form a consumer group.
        subscription = KubeMQ::PubSub::EventsStoreSubscription.new(
          channel: 'orders.processing',
          group: 'order-processors',
          start_position: KubeMQ::PubSub::EventStoreStartPosition::START_FROM_FIRST
        )

        client.subscribe_to_events_store(subscription, cancellation_token: cancel,
                                         on_error: ->(e) { puts "[#{worker_id}] Error: #{e.message}" }) do |event|
          puts "[#{worker_id}] Processing seq=#{event.sequence}: #{event.body}"
        end

        puts "[#{worker_id}] Ready in group 'order-processors'"
        sleep 300
        ensure
        cancel&.cancel
        client&.close
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="order_worker.exs"
        worker_id = System.get_env("WORKER_ID", "worker-1")

        {:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: worker_id)

        # Subscribers sharing the same group name form a consumer group.
        {:ok, _sub} =
          KubeMQ.Client.subscribe_to_events_store(client, "orders.processing",
            start_at: :start_from_first,
            group: "order-processors",
            on_event: fn event ->
              IO.puts("[#{worker_id}] Processing seq=#{event.sequence}: #{event.body}")
            end,
            on_error: fn err -> IO.puts("[#{worker_id}] Error: #{err}") end
          )

        IO.puts("[#{worker_id}] Ready in group 'order-processors'")
        Process.sleep(300_000)
        ```
      </Tab>
    </Tabs>

    Run multiple instances with different `WORKER_ID` values:

    ```bash
    WORKER_ID=worker-A go run order_worker.go &
    WORKER_ID=worker-B go run order_worker.go &
    WORKER_ID=worker-C go run order_worker.go &
    ```
  </Step>

  <Step>
    ### Publish Events and Verify Distribution [#publish-events-and-verify-distribution]

    Publish 6 order events and observe round-robin distribution across the 3 workers.

    ```text
    [worker-A] Processing seq=1: {"orderId":"ORD-1001",...}
    [worker-B] Processing seq=2: {"orderId":"ORD-1002",...}
    [worker-C] Processing seq=3: {"orderId":"ORD-1003",...}
    [worker-A] Processing seq=4: {"orderId":"ORD-1004",...}
    [worker-B] Processing seq=5: {"orderId":"ORD-1005",...}
    [worker-C] Processing seq=6: {"orderId":"ORD-1006",...}
    ```

    Each event is delivered to exactly one worker. The workload is distributed evenly.
  </Step>

  <Step>
    ### Verify Durable Resume After Disconnect [#verify-durable-resume-after-disconnect]

    1. Workers A, B, C process events up to sequence 100
    2. All three workers disconnect
    3. 50 new events arrive (seq 101-150)
    4. Worker A reconnects with the same group name
    5. Worker A receives events starting from sequence 101

    The `StartPosition` parameter is only used on the **first connection** for a durable name. Subsequent connections resume from the last tracked position.
  </Step>
</Steps>

## Groups vs Fan-Out [#groups-vs-fan-out]

<Mermaid
  chart="graph TD
  subgraph FanOut[&#x22;Fan-Out (no group)&#x22;]
    P1[&#x22;Publisher&#x22;]
    CH1{{&#x22;Channel&#x22;}}
    S1A[&#x22;Sub A — gets ALL events&#x22;]
    S1B[&#x22;Sub B — gets ALL events&#x22;]
    P1 --> CH1
    CH1 --> S1A
    CH1 --> S1B
  end
  subgraph GroupLB[&#x22;Consumer Group&#x22;]
    P2[&#x22;Publisher&#x22;]
    CH2{{&#x22;Channel&#x22;}}
    S2A[&#x22;Sub A — gets 50%&#x22;]
    S2B[&#x22;Sub B — gets 50%&#x22;]
    P2 --> CH2
    CH2 -- &#x22;round-robin&#x22; --> S2A
    CH2 -- &#x22;round-robin&#x22; --> S2B
  end

  class CH1 events
  class CH2 queue
  class P1,P2,S1A,S1B,S2A,S2B client"
/>

*Without a group every subscriber gets a full copy; within a group the channel load-balances each event to one member.*

| Delivery          | No Group                                  | With Group                    |
| ----------------- | ----------------------------------------- | ----------------------------- |
| Event routing     | Every subscriber gets every event         | Each event goes to one member |
| Use case          | Independent processing (audit, analytics) | Load-balanced processing      |
| Position tracking | Per subscriber                            | Per group (shared)            |

## Multiple Groups on One Channel [#multiple-groups-on-one-channel]

Different groups receive independent copies of the event stream:

```bash
# Group 1: Order fulfillment (3 workers sharing load)
WORKER_ID=fulfill-1 GROUP=fulfillment go run worker.go
WORKER_ID=fulfill-2 GROUP=fulfillment go run worker.go

# Group 2: Analytics (2 workers sharing load)
WORKER_ID=analytics-1 GROUP=analytics go run worker.go
WORKER_ID=analytics-2 GROUP=analytics go run worker.go

# No group: Auditor (receives every event)
go run auditor.go
```

Each group independently tracks its position and distributes events among its members.

## Events Store Groups vs Events Groups [#events-store-groups-vs-events-groups]

| Feature            | Events Groups            | Events Store Groups             |
| ------------------ | ------------------------ | ------------------------------- |
| Position tracking  | None (ephemeral)         | Durable (survives disconnect)   |
| Missed messages    | Lost when offline        | Replayed on reconnect           |
| Delivery guarantee | At-most-once             | At-least-once                   |
| Replay capability  | None                     | Full history replay             |
| Use case           | Real-time load balancing | Reliable distributed processing |

<Callout type="info">
  To force a fresh replay, use a different group name. The old group's position data remains until the channel is purged or the inactive purge timeout expires.
</Callout>

## Scaling Guidelines [#scaling-guidelines]

| Factor            | Recommendation                                                      |
| ----------------- | ------------------------------------------------------------------- |
| Number of members | Scale based on processing throughput. No hard limit.                |
| Slow consumers    | Keep processing fast or offload to background workers.              |
| Group naming      | Use descriptive names (e.g., `email-senders`, `report-generators`). |
| Rebalancing       | Adding or removing group members takes effect immediately.          |

## Next Steps [#next-steps]

* Learn about [stream publishing](/learn/events-store/tutorials/stream-publishing) for high throughput
* Configure [retention policies](/learn/events-store/how-to/configure-retention)
* Handle [resume after disconnect](/learn/events-store/how-to/resume-after-disconnect)
* See the [Events Store Reference](/learn/events-store/reference) for subscription parameters