Query Group

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: passing group in 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
• Node.js SDK installed (npm install kubemq-js)

Code

import { KubeMQClient, createQuery } from 'kubemq-js';

async function main() {
  const client = await KubeMQClient.create({
    address: 'localhost:50000',
    clientId: 'js-rpc-query-group-client',
  });

  try {
    const sub = client.subscribeToQueries({
      channel: 'js-rpc.query-group',
      group: 'responders',
      onQuery: async (q) => {
        console.log('Group handler received query:', q.id);
        await client.sendQueryResponse({
          id: q.id,
          replyChannel: q.replyChannel,
          executed: true,
          body: new TextEncoder().encode(JSON.stringify({ answer: 42 })),
        });
      },
      onError: (err) => {
        console.error('Sub error:', err.message);
      },
    });

    await new Promise((r) => setTimeout(r, 500));
    const resp = await client.sendQuery(
      createQuery({ channel: 'js-rpc.query-group', body: 'question', timeoutInSeconds: 5 }),
    );
    console.log('Query executed:', resp.executed);
    if (resp.body) console.log('Response body:', new TextDecoder().decode(resp.body));

    sub.cancel();
  } finally {
    await client.close();
  }
}

main().catch(console.error);

How It Works

• group: 'responders' in subscribeToQueries() places this handler in the responders consumer group; KubeMQ routes each query to exactly one group member for load balancing.
• The 500 ms wait after subscribing ensures the gRPC subscription stream is registered before the sender issues the query — preventing a race condition.
• await client.sendQueryResponse() in the async (q) => ... callback correctly awaits the response send; the async callback signature is required to propagate the awaited response before any timeout.
• Scale out by running multiple processes with the same group: 'responders' — KubeMQ distributes queries across all active members without duplication.

Related

• Pattern overview
• Node.js SDK Reference
• Send Query
• Handle Query