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

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)

Code

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

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.

Query Group

Overview

Prerequisites

Code

How It Works

On this page

Query Group

Overview

Prerequisites

Code

How It Works

Related

On this page