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

KubeMQ server running on localhost:50000
Kotlin SDK installed (implementation("io.kubemq.sdk:kubemq-sdk-kotlin:1.0.1"))

Code

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

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.

Consumer Group

Overview

Prerequisites

Code

How It Works

On this page

Consumer Group

Overview

Prerequisites

Code

How It Works

Related

On this page