# Query Group (/sdks/java/how-to/rpc/query-group)



## Overview [#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: `QueriesSubscription.builder().channel(...).group(...)` passed to `client.subscribeToQueries` load-balances across every subscription 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 [#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.queries;

import io.kubemq.sdk.cq.*;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

public class ConsumerGroupExample {
    private static final String ADDRESS = "localhost:50000";
    private static final String CLIENT_ID = "java-queries-consumer-group-client";
    private static final String CHANNEL = "java-queries.consumer-group";

    public static void main(String[] args) throws InterruptedException {
        // Create a client connected to the KubeMQ server
        CQClient client = CQClient.builder().address(ADDRESS).clientId(CLIENT_ID).build();
        client.ping();
        // Create the queries channel
        client.createQueriesChannel(CHANNEL);

        // Subscribe multiple handlers in a consumer group (load-balanced)
        String group = "query-handlers";
        AtomicInteger[] counts = new AtomicInteger[3];
        QueriesSubscription[] subs = new QueriesSubscription[3];

        for (int i = 0; i < 3; i++) {
            final int id = i + 1;
            counts[i] = new AtomicInteger(0);
            final AtomicInteger counter = counts[i];
            Map<String, String> respTags = new HashMap<>();
            respTags.put("handler", String.valueOf(id));

            subs[i] = QueriesSubscription.builder()
                    .channel(CHANNEL).group(group)
                    .onReceiveQueryCallback(q -> {
                        counter.incrementAndGet();
                        client.sendResponseMessage(QueryResponseMessage.builder()
                                .queryReceived(q).isExecuted(true)
                                .body(("Handler " + id).getBytes()).tags(respTags).build());
                    })
                    .onErrorCallback(err -> {}).build();
            // Subscribe each handler to the consumer group
            client.subscribeToQueries(subs[i]);
        }
        Thread.sleep(500);

        // Send queries to the group (distributed across handlers)
        System.out.println("Sending 9 queries to group '" + group + "'...\n");
        for (int i = 1; i <= 9; i++) {
            try {
                QueryResponseMessage r = client.sendQuery(QueryMessage.builder()
                        .channel(CHANNEL).body(("Query #" + i).getBytes()).timeoutInSeconds(10).build());
                System.out.println("  Query #" + i + " -> " + r.getTags().get("handler"));
            } catch (Exception e) { /* handle */ }
        }

        System.out.println("\nDistribution:");
        for (int i = 0; i < 3; i++) { System.out.println("  Handler " + (i + 1) + ": " + counts[i].get()); }

        // Clean up resources
        for (QueriesSubscription s : subs) { s.cancel(); }
        client.deleteQueriesChannel(CHANNEL);
        client.close();
    }
}

```

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

* Three `QueriesSubscription` objects share the same `group("query-handlers")` on the same channel; KubeMQ delivers each incoming query to exactly one handler in the group (round-robin or least-busy), providing load balancing.
* Each handler tags its response with `"handler": id` so the caller can log which handler processed each query.
* `sendQuery()` blocks until any one handler in the group responds; from the caller's perspective, groups are transparent — it targets the channel, not a specific handler.
* In production each handler would run in a separate JVM/process; here all three share one `CQClient` for simplicity.

## Related [#related]

* [Pattern overview](/learn/rpc/getting-started)
* [Java SDK Reference](/sdks/java/reference/rpc)
* [Send Query](/sdks/java/tutorials/query-send)
* [Handle Query](/sdks/java/how-to/rpc/query-handle)