Consumer Group

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 EventsStoreSubscription::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
• Rust SDK installed (cargo add kubemq)

Code

use kubemq::prelude::*;
use kubemq::{EventStoreBuilder, EventsStoreSubscription};
use std::time::Duration;

#[tokio::main]
async fn main() -> kubemq::Result<()> {
    let client = KubemqClient::builder()
        .host("localhost")
        .port(50000)
        .build()
        .await?;

    let channel = "rust-events-store.consumer-group";
    let group = "store-consumer-group";

    let sub1 = client.subscribe_to_events_store(
        channel, group,
        EventsStoreSubscription::StartFromFirst,
        |event| Box::pin(async move {
            println!("Consumer-1: seq={}, body={}", event.sequence, String::from_utf8_lossy(&event.body));
        }),
        None,
    ).await?;

    let sub2 = client.subscribe_to_events_store(
        channel, group,
        EventsStoreSubscription::StartFromFirst,
        |event| Box::pin(async move {
            println!("Consumer-2: seq={}, body={}", event.sequence, String::from_utf8_lossy(&event.body));
        }),
        None,
    ).await?;

    tokio::time::sleep(Duration::from_millis(500)).await;

    for i in 0..5 {
        let event = EventStoreBuilder::new()
            .channel(channel)
            .body(format!("stored-msg-{}", i).into_bytes())
            .build();
        client.send_event_store(event).await?;
    }

    tokio::time::sleep(Duration::from_secs(2)).await;
    sub1.unsubscribe().await;
    sub2.unsubscribe().await;
    client.close().await?;
    Ok(())
}

How It Works

• Both subscribers pass the same group name for load-balanced delivery.
• Each stored event is delivered to exactly one group member.
• Review timeouts, channel names, and client IDs before running against shared environments.
• Run the program while the server from the prerequisites is available.

Related

• Events Store overview
• Rust SDK Reference
• Persistent Pub/Sub