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.

For the underlying concept — why a consumer group gives you load-balancing and broadcast from the same channel — see Scaling & flow in the Fundamentals track. This page focuses on the Events Store specifics: durable position tracking and resume after disconnect.

How Consumer Groups Work

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

KubeMQ server running on localhost:50000
SDK installed (Getting Started)

Step-by-Step

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.

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()
}

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)

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'`);

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();

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)}");
}

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)}")
    }
}

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;
    });

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(())
}

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

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)

Run multiple instances with different WORKER_ID values:

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 &

Publish Events and Verify Distribution

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

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

Verify Durable Resume After Disconnect

Workers A, B, C process events up to sequence 100
All three workers disconnect
50 new events arrive (seq 101-150)
Worker A reconnects with the same group name
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.

Groups vs Fan-Out

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

Different groups receive independent copies of the event stream:

# 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

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

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.

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

Learn about stream publishing for high throughput
Configure retention policies
Handle resume after disconnect
See the Events Store Reference for subscription parameters

Durable Consumer Groups