Overview

A consumer group scales query handling horizontally without touching the caller's side. Instead of one process answering every query on a channel, you run several identical handler instances under the same group name, and the broker routes each query to exactly one member — never to all of them. That turns a single responder into a pool you can grow or shrink to match load, which matters for anything RPC-shaped: a lookup service, a cache-fill handler, a synchronous read path behind an API.

It works by tying group membership to the subscription: setting group inside client.subscribeToQueries { channel = ...; group = ... } load-balances across every subscriber sharing that channel and group. The sender calls client.sendQuery exactly as it would against a single handler — it never knows how many members exist or which one answered.

Gotchas: channel and group name must match exactly, or a typo quietly creates a second, empty group instead of erroring. Omit group and every subscriber reverts to broadcast, each answering independently. A stuck group member isn't bypassed — the caller just sees a timeout.

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.queries

import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.cq.queryMessage
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-queries-query-group"
private const val CHANNEL = "kotlin-queries.query-group"

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

    client.use {
        try {
            client.createQueriesChannel(CHANNEL)

            // 3 handlers in the same consumer group
            val group = "query-handlers"
            val numWorkers = 3
            val counts = Array(numWorkers) { AtomicInteger(0) }

            val jobs = (0 until numWorkers).map { i ->
                launch {
                    client.subscribeToQueries {
                        channel = CHANNEL
                        this.group = group
                    }.collect { query ->
                        counts[i].incrementAndGet()
                        val response = query.respond {
                            executed = true
                            body = "Response from worker ${i + 1}".toByteArray()
                            tags = mapOf("worker" to "${i + 1}")
                        }
                        client.sendQueryResponse(response)
                    }
                }
            }

            delay(500)

            // Send 9 queries
            println("Sending 9 queries to group '$group'...\n")
            repeat(9) { i ->
                try {
                    val resp = client.sendQuery(queryMessage {
                        channel = CHANNEL
                        body = "Query #${i + 1}".toByteArray()
                        timeoutMs = 10000
                    })
                    println("Query ${i + 1}: handled by worker ${resp.tags["worker"]}")
                } catch (e: Exception) {
                    println("Query ${i + 1} error: ${e.message}")
                }
            }

            delay(500)

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

            jobs.forEach { it.cancel() }
        } finally {
            try { client.deleteQueriesChannel(CHANNEL) } catch (_: Exception) {}
        }
        println("Done.")
    }
}

How It Works

Setting group = "query-handlers" distributes queries across group members.
Each query is handled by exactly one worker in the group.
Response tags identify which worker handled each query.
Consumer groups work the same way for commands and queries.

Query Group

Overview

Prerequisites

Code

How It Works

On this page

Query Group

Overview

Prerequisites

Code

How It Works

Related

On this page