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 to subscribe_to_events_store on each subscriber alongside a start position such as EventStoreStartPosition.StartFromFirst. 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

KubeMQ server running on localhost:50000
Python SDK installed (pip install kubemq)

Code

consumer_group.py

"""Example: Consumer group — load-balance events store messages across subscribers."""

from __future__ import annotations

import asyncio

from kubemq import AsyncCancellationToken, AsyncPubSubClient, EventStoreMessage, EventsStoreSubscription
from kubemq.pubsub import EventStoreStartPosition


async def main() -> None:
    async with AsyncPubSubClient(
        address="localhost:50000",
        client_id="python-events-store-consumer-group-client",
    ) as client:
        token = AsyncCancellationToken()

        async def make_processor(name: str) -> None:
            async for event in client.subscribe_to_events_store(
                subscription=EventsStoreSubscription(
                    channel="python-events-store.consumer-group",
                    group="processors",
                    on_receive_event_callback=lambda e: None,
                    on_error_callback=lambda e: print(f"Error: {e}"),
                    events_store_type=EventStoreStartPosition.StartFromFirst,
                ),
                cancellation_token=token,
            ):
                print(f"[{name}] Seq:{event.sequence}, Body:{event.body.decode('utf-8')}")

        task1 = asyncio.create_task(make_processor("Processor-1"))
        task2 = asyncio.create_task(make_processor("Processor-2"))
        await asyncio.sleep(1)

        for i in range(6):
            await client.send_event_store(
                EventStoreMessage(
                    channel="python-events-store.consumer-group",
                    body=f"Event-{i + 1}".encode(),
                )
            )

        await asyncio.sleep(3)
        token.cancel()
        for t in [task1, task2]:
            t.cancel()
            try:
                await t
            except asyncio.CancelledError:
                pass


if __name__ == "__main__":
    asyncio.run(main())

How It Works

Both processors subscribe with group="processors" on the same channel; KubeMQ routes each stored event to exactly one processor in the group.
EventStoreStartPosition.StartFromFirst means both processors begin by replaying all historical events, distributed between them.
Consumer groups on events-store channels provide exactly-once delivery per group, enabling parallel processing without duplicate work.
A single AsyncCancellationToken stops both processor tasks simultaneously when cancel() is called.

Consumer Group

Overview

Prerequisites

Code

How It Works

On this page

Consumer Group

Overview

Prerequisites

Code

How It Works

Related

On this page