# Consumer Group (/sdks/cpp/how-to/events-store/consumer-group)



## Overview [#overview]

A **consumer group** turns Events Store from a broadcast fan-out into a competing-consumers queue: subscribers sharing the same group split the stored events between them instead of each getting a copy of every event. Reach for this when a durable, ordered event log also needs to scale horizontally — a stream of order updates or audit records where one processor can't keep up, but each event still needs to be handled exactly once by the group as a whole.

It works by passing the same `group` string to `SubscribeToEventsStore` on each subscriber alongside a start position such as `kubemq::SubscriptionOption::StartFromFirstEvent()`. The broker load-balances deliveries across every active member sharing that group and channel; adding another subscriber with the same group name is all it takes to add capacity. **Gotchas:** the start position belongs to the group's shared read cursor, not to any one subscriber — members joining later pick up wherever the group already is, not from the beginning. Different group names silently mean broadcast instead of load balancing, with no error to warn you. Delivery is exactly-once per group, but a crashed member's in-flight event isn't automatically handed to another member — design processing to be safely restartable.

## Prerequisites [#prerequisites]

* KubeMQ server running on `localhost:50000`
* C++ SDK installed (vcpkg or CMake FetchContent)
* C++17 compiler (GCC 9+, Clang 9+, MSVC 2019+)

## Code [#code]

```cpp title="main.cc"
// Example: events_store/consumer_group
//
// Demonstrates load-balanced event store consumption with consumer groups.
// When multiple subscribers share the same group, each event is delivered
// to exactly one member.
//
// Channel: cpp-events-store.consumer-group
// Client ID: cpp-events-store-consumer-group-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).

#include <kubemq/kubemq.h>

#include <atomic>
#include <chrono>
#include <iostream>
#include <thread>

int main() {
    std::cout << "[1] Connecting to localhost:50000" << std::endl;

    kubemq::ClientOptions options;
    options.set_address("localhost", 50000);
    options.set_client_id("cpp-events-store-consumer-group-client");

    auto client_result = kubemq::Client::Create(options);
    if (!client_result.ok()) {
        std::cerr << "[ERROR] Failed to create client: " << client_result.status().message()
                  << std::endl;
        return 1;
    }
    auto& client = *client_result;

    std::string channel = "cpp-events-store.consumer-group";
    std::string group = "cpp-events-store-worker-group";
    std::atomic<int> received_count{0};

    // Subscribe with a consumer group for load-balanced event store delivery.
    std::cout << "[2] Subscribing with group=" << group << std::endl;
    auto sub_result = client->SubscribeToEventsStore(
        channel, group, kubemq::SubscriptionOption::StartFromFirstEvent(),
        [&received_count](const kubemq::EventStoreReceive& e) {
            std::cout << "[4] [ConsumerGroup] seq=" << e.sequence << " body=" << e.body
                      << std::endl;
            received_count.fetch_add(1);
        },
        [](const kubemq::Status& err) {
            std::cerr << "[ERROR] Subscription error: " << err.message() << std::endl;
        });
    if (!sub_result.ok()) {
        std::cerr << "[ERROR] SubscribeToEventsStore: " << sub_result.status().message()
                  << std::endl;
        return 1;
    }
    auto& sub = *sub_result;

    // Allow subscription to establish.
    std::this_thread::sleep_for(std::chrono::seconds(1));

    // Send an event.
    std::cout << "[3] Sending event to channel " << channel << std::endl;
    auto ev_result = kubemq::EventStore::Builder()
                         .SetChannel(channel)
                         .SetBody("group event")
                         .SetMetadata("group-demo")
                         .Build();
    if (!ev_result.ok()) {
        std::cerr << "[ERROR] Build: " << ev_result.status().message() << std::endl;
        return 1;
    }
    auto send_result = client->SendEventStore(*ev_result);
    if (!send_result.ok()) {
        std::cerr << "[ERROR] SendEventStore: " << send_result.status().message() << std::endl;
        return 1;
    }

    // Wait for the event to be received.
    std::this_thread::sleep_for(std::chrono::seconds(2));
    std::cout << "[5] Received " << received_count.load() << " event(s)" << std::endl;
    std::cout << "[6] Consumer group demo complete" << std::endl;

    // Cancel the subscription explicitly.
    // Note: The Subscription destructor also calls Cancel(), but explicit
    // cleanup is shown here for clarity and to match Go's defer pattern.
    sub->Cancel();

    auto close_status = client->Close();
    if (!close_status.ok()) {
        std::cerr << "[ERROR] Close failed: " << close_status.message() << std::endl;
        return 1;
    }
    std::cout << "[7] Client closed" << std::endl;

    return 0;
}
```

## How It Works [#how-it-works]

* Subscribes with a consumer group name for load-balanced delivery.
* Uses `StartFromFirstEvent()` to replay all stored events within the group.
* Each event is delivered to exactly one member of the consumer group.
* Enables horizontal scaling of persistent event processors.

## Related [#related]

* [Pattern overview](/learn/events-store/getting-started)
* [C++ SDK Reference](/sdks/cpp/reference/events-store)
* [Persistent Pub/Sub](/sdks/cpp/tutorials/persistent-pubsub)
* [Events Consumer Group](/sdks/cpp/how-to/events/consumer-group)