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 setting the same .group(groupName) on each EventsStoreSubscription passed to subscribeToEventsStore, alongside a start position such as EventsStoreType.StartNewOnly. The broker load-balances deliveries across every active member sharing that group and channel; adding another subscription 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
Java SDK installed (implementation 'io.kubemq.sdk:kubemq-sdk-Java:3.1.1' (Gradle) or Maven dependency from Getting Started)

Code

ConsumerGroupExample.java

package io.kubemq.example.eventsstore;

import io.kubemq.sdk.pubsub.*;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Consumer Group Example (EventsStore)
 *
 * Demonstrates load-balanced EventsStore consumption using subscription groups.
 */
public class ConsumerGroupExample {

    private static final String ADDRESS = "localhost:50000";
    private static final String CLIENT_ID = "java-eventsstore-consumer-group-client";
    private static final String CHANNEL = "java-eventsstore.consumer-group";

    public static void main(String[] args) throws InterruptedException {
        // Create a client connected to the KubeMQ server
        PubSubClient client = PubSubClient.builder().address(ADDRESS).clientId(CLIENT_ID).build();
        client.ping();
        // Create the events store channel
        client.createEventsStoreChannel(CHANNEL);

        String groupName = "store-processors";
        AtomicInteger proc1 = new AtomicInteger(0);
        AtomicInteger proc2 = new AtomicInteger(0);
        CountDownLatch latch = new CountDownLatch(6);

        EventsStoreSubscription sub1 = EventsStoreSubscription.builder()
                .channel(CHANNEL).group(groupName)
                .eventsStoreType(EventsStoreType.StartNewOnly)
                .onReceiveEventCallback(e -> { proc1.incrementAndGet(); System.out.println("  Processor 1: " + new String(e.getBody())); latch.countDown(); })
                .onErrorCallback(err -> {}).build();

        EventsStoreSubscription sub2 = EventsStoreSubscription.builder()
                .channel(CHANNEL).group(groupName)
                .eventsStoreType(EventsStoreType.StartNewOnly)
                .onReceiveEventCallback(e -> { proc2.incrementAndGet(); System.out.println("  Processor 2: " + new String(e.getBody())); latch.countDown(); })
                .onErrorCallback(err -> {}).build();

        // Subscribe two processors in the same group (load-balanced)
        client.subscribeToEventsStore(sub1);
        client.subscribeToEventsStore(sub2);
        System.out.println("Two processors in group: " + groupName);
        Thread.sleep(500);

        // Send events (distributed across processors in the group)
        for (int i = 1; i <= 6; i++) {
            client.publishEventStore(EventStoreMessage.builder()
                    .id("event-" + i).channel(CHANNEL)
                    .body(("Event #" + i).getBytes()).build());
            Thread.sleep(100);
        }

        latch.await(5, TimeUnit.SECONDS);
        System.out.println("\nProcessor 1: " + proc1.get() + ", Processor 2: " + proc2.get());

        // Clean up resources
        sub1.cancel(); sub2.cancel();
        client.deleteEventsStoreChannel(CHANNEL);
        client.close();
    }
}

How It Works

Both subscriptions use .group("store-processors") and StartNewOnly; the broker delivers each stored event to exactly one processor within the group.
StartNewOnly means neither processor replays historical events — they compete only over events published after subscription time.
Per-processor AtomicInteger counters printed at the end show the load distribution; with six events and two processors the split is approximately equal.

Consumer Group

Overview

Prerequisites

Code

How It Works

Related

On this page