# Consumer Group (/sdks/java/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 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 [#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](/sdks/java))

## Code [#code]

```java title="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 [#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.

## Related [#related]

* [Pattern overview](/learn/events-store/getting-started)
* [Java SDK Reference](/sdks/java/reference/events-store)
* [Persistent Pub/Sub](/sdks/java/tutorials/persistent-pubsub)
* [Cancel Subscription](/sdks/java/how-to/events-store/cancel-subscription)