# Command Group (/sdks/kotlin/how-to/rpc/command-group)



## Overview [#overview]

A command **consumer group** turns a single command handler into a scalable worker pool: run multiple identical instances subscribed with the same group name, and the broker load-balances each incoming command to exactly one member instead of broadcasting it to all of them. This is how you add capacity to handle a growing command volume — start more workers in the same group — without changing anything on the caller's side.

Every subscriber sets the same `group` alongside `channel` in its `subscribeToCommands` builder; the broker tracks membership and picks one live member per command. `sendCommand` on the caller side is unaware groups exist — it just suspends for a response, which comes back from whichever worker happened to handle it.

**Gotchas:** group membership is scoped per channel — subscribers on the same channel with *different* group names each get their own full copy of every command (fan-out), which looks like a bug when you expected load-balancing. A slow handler still holds up the caller's timeout, since only one worker is ever picked. And if every member of the group is offline when a command arrives, the send simply fails or times out — commands aren't queued or replayed for a group that has no active listener.

## 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.commands

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

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

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

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

            val jobs = (0 until numWorkers).map { i ->
                launch {
                    client.subscribeToCommands {
                        channel = CHANNEL
                        this.group = group
                    }.collect { cmd ->
                        counts[i].incrementAndGet()
                        val response = cmd.respond { executed = true }
                        client.sendCommandResponse(response)
                    }
                }
            }

            delay(500)

            // Send 9 commands to the group (distributed across workers)
            println("Sending 9 commands to group '$group'...\n")
            repeat(9) { i ->
                try {
                    client.sendCommand(commandMessage {
                        channel = CHANNEL
                        body = "Cmd #${i + 1}".toByteArray()
                        timeoutMs = 10000
                    })
                } catch (e: Exception) {
                    println("Command ${i + 1} error: ${e.message}")
                }
            }

            delay(1000)

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

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

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

* Setting `group = "command-handlers"` enables consumer group mode for command subscriptions.
* Each command is delivered to exactly one worker in the group.
* The server distributes commands across group members for load balancing.
* All workers share the same group name but handle different commands.

## Related [#related]

* [Send Command](/sdks/kotlin/tutorials/command-send)
* [Query Group](/sdks/kotlin/how-to/rpc/query-group)
* [RPC Reference](/sdks/kotlin/reference/rpc)