# Consumer Group (/sdks/kotlin/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` in the `subscribeToEventsStore` builder for each subscriber, alongside a `startPosition` such as `StartPosition.StartNewOnly`. 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`
* Kotlin SDK installed (`implementation("io.kubemq.sdk:kubemq-sdk-kotlin:1.0.1")`)

## Code [#code]

```kotlin title="ConsumerGroupExample.kt"
package io.kubemq.sdk.examples.eventsstore

import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.pubsub.StartPosition
import io.kubemq.sdk.pubsub.eventStoreMessage
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import java.util.concurrent.atomic.AtomicInteger

private const val ADDRESS = "localhost:50000"
private const val CLIENT_ID = "kotlin-events-store-consumer-group"
private const val CHANNEL = "kotlin-events-store.consumer-group"

fun main() = runBlocking {
    val client = KubeMQClient.pubSub {
        address = ADDRESS
        clientId = CLIENT_ID
    }

    client.use {
        val numSubscribers = 2
        val numMessages = 6
        val counts = Array(numSubscribers) { AtomicInteger(0) }
        val totalReceived = AtomicInteger(0)

        // Subscribe with consumer group
        val jobs = (0 until numSubscribers).map { i ->
            launch {
                client.subscribeToEventsStore {
                    channel = CHANNEL
                    group = "store-processors"
                    startPosition = StartPosition.StartNewOnly
                }.collect { msg ->
                    counts[i].incrementAndGet()
                    println("[group-${i + 1}] seq=${msg.sequence}: ${String(msg.body)}")
                    totalReceived.incrementAndGet()
                }
            }
        }

        delay(500)

        // Publish persistent events
        repeat(numMessages) { i ->
            client.publishEventStore(eventStoreMessage {
                channel = CHANNEL
                body = "Store event #${i + 1}".toByteArray()
            })
            delay(100)
        }

        delay(2000)

        println("\nDistribution:")
        for (i in 0 until numSubscribers) {
            println("  Subscriber ${i + 1}: ${counts[i].get()} messages")
        }

        jobs.forEach { it.cancel() }
        println("Done.")
    }
}
```

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

* Setting `group = "store-processors"` enables consumer group mode for events store.
* Each persistent event is delivered to exactly one member of the group.
* The `startPosition` is shared across group members.
* Consumer groups work the same way for events and events store subscriptions.

## Related [#related]

* [Persistent Pub/Sub](/sdks/kotlin/tutorials/persistent-pubsub)
* [Events Consumer Group](/sdks/kotlin/how-to/events/consumer-group)
* [Events Store Reference](/sdks/kotlin/reference/events-store)