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: to KubeMQ.Client.subscribe_to_queries(client, channel, group: ..., on_query: ...) load-balances across every subscriber sharing that channel and group. The sender calls KubeMQ.Client.send_query 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
• Elixir SDK installed ({:kubemq, "~> 1.0"} in mix.exs)

Code

channel = "elixir-rpc.query-group"
group = "query-workers"
{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "elixir-query-group")

{:ok, sub1} =
  KubeMQ.Client.subscribe_to_queries(client, channel,
    group: group,
    on_query: fn query ->
      IO.puts("[Worker-1] Handling: #{query.body}")
      KubeMQ.QueryReply.new(
        request_id: query.id,
        response_to: query.reply_channel,
        executed: true,
        body: "Response from Worker-1"
      )
    end
  )

{:ok, sub2} =
  KubeMQ.Client.subscribe_to_queries(client, channel,
    group: group,
    on_query: fn query ->
      IO.puts("[Worker-2] Handling: #{query.body}")
      KubeMQ.QueryReply.new(
        request_id: query.id,
        response_to: query.reply_channel,
        executed: true,
        body: "Response from Worker-2"
      )
    end
  )

IO.puts("Two query workers in group '#{group}' ready")
Process.sleep(500)

for i <- 1..4 do
  query = KubeMQ.Query.new(channel: channel, body: "Query #{i}", timeout: 10_000)

  case KubeMQ.Client.send_query(client, query) do
    {:ok, resp} -> IO.puts("Query #{i} → #{resp.body}")
    {:error, err} -> IO.puts("Query #{i} failed: #{err.message}")
  end
end

KubeMQ.Subscription.cancel(sub1)
KubeMQ.Subscription.cancel(sub2)
KubeMQ.Client.close(client)

How It Works

• The group option distributes queries across handlers in the same consumer group
• Each query is routed to exactly one handler for processing
• The response body indicates which worker processed each query

Related

• Send Query
• Command Group