Commands and Queries (CQ) is KubeMQ's synchronous request-response pattern. Unlike fire-and-forget events, every CQ request blocks until the matching handler replies (or the timeout elapses), giving you RPC-style semantics over the message broker.

The two halves of the pattern differ in what the response carries:

Commands signal success or failure — the response is a boolean isExecuted flag (optionally with an error). Use commands to trigger an action: restart a service, deploy a build, run a job.
Queries return a data payload — the response carries a body alongside isExecuted. Use queries to fetch data: look up a user, read a configuration, compute a value.

CQ is point-to-point: each request is routed to exactly one handler (or, with a consumer group, one member of the group). This is why the handler annotations take a singular channel — a CQ handler services exactly one channel.

This page documents the Spring API for Commands and Queries. For the underlying request-response model, see the core Commands & Queries (RPC) concept.

A command or query blocks on the template until the point-to-point handler replies (or the timeout elapses).

The Spring Boot starter speaks gRPC to the broker directly on port 50000 — CQ does not use the shared HTTP connector. Start a local broker with Docker (the gRPC port 50000 is all CQ needs; 9090 exposes the shared HTTP/REST and dashboard endpoints):

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

Sending Commands

Inject KubeMQTemplate and call sendCommand(channel, data, Duration timeout). The call blocks until the handler replies or the timeout elapses, then returns a CommandResponseMessage whose isExecuted() flag tells you whether the handler succeeded.

SendCommandRunner.java

package io.kubemq.spring.boot.examples.commands;

import io.kubemq.sdk.cq.CommandResponseMessage;
import io.kubemq.spring.boot.autoconfigure.template.KubeMQTemplate;
import java.time.Duration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

@Component
public class SendCommandRunner implements ApplicationRunner {

    private static final Logger log = LoggerFactory.getLogger(SendCommandRunner.class);

    private final KubeMQTemplate template;

    public SendCommandRunner(KubeMQTemplate template) {
        this.template = template;
    }

    @Override
    public void run(ApplicationArguments args) {
        CommandResponseMessage response = template.sendCommand(
                "spring-commands.send", "Restart service", Duration.ofSeconds(30));
        log.info("Command response: executed={}", response.isExecuted());
    }
}

The third argument is the send-side timeout — how long the caller waits for the handler's reply. Pass it as a java.time.Duration; the template converts it to whole seconds internally (see Timeouts).

For non-blocking sends, sendCommandAsync(channel, data, timeout) returns a CompletableFuture<CommandResponseMessage> so you can compose the result without blocking the calling thread:

Async command

template.sendCommandAsync("spring-commands.send", "Restart service", Duration.ofSeconds(30))
        .thenAccept(response -> log.info("Command executed={}", response.isExecuted()));

Handling Commands

Annotate a bean method with @KubeMQCommandHandler(channel = ...). The method receives a CommandMessageReceived and returns either a boolean (simple mode) or a CommandResponseMessage (full control). Because CQ is point-to-point, the annotation takes a singular channel — each handler services exactly one command channel.

Simple mode — return a boolean

Returning true reports the command as executed; false reports failure. This is the most concise form when you only need to signal success.

SendCommandHandler.java

package io.kubemq.spring.boot.examples.commands;

import io.kubemq.sdk.cq.CommandMessageReceived;
import io.kubemq.spring.boot.autoconfigure.listener.KubeMQCommandHandler;
import java.nio.charset.StandardCharsets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

@Component
public class SendCommandHandler {

    private static final Logger log = LoggerFactory.getLogger(SendCommandHandler.class);

    @KubeMQCommandHandler(channel = "spring-commands.send")
    public boolean onCommand(CommandMessageReceived cmd) {
        String body = cmd.getBody() != null
                ? new String(cmd.getBody(), StandardCharsets.UTF_8) : "<empty>";
        log.info("Command received: channel={} body={}", cmd.getChannel(), body);
        return true; // executed successfully
    }
}

Full control — return a CommandResponseMessage

When you need to correlate the response explicitly or set the executed flag conditionally, return a CommandResponseMessage built with .commandReceived(cmd) (which ties the response to the originating request) and .isExecuted(...).

HandleCommandHandler.java

package io.kubemq.spring.boot.examples.commands;

import io.kubemq.sdk.cq.CommandMessageReceived;
import io.kubemq.sdk.cq.CommandResponseMessage;
import io.kubemq.spring.boot.autoconfigure.listener.KubeMQCommandHandler;
import java.nio.charset.StandardCharsets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

@Component
public class HandleCommandHandler {

    private static final Logger log = LoggerFactory.getLogger(HandleCommandHandler.class);

    @KubeMQCommandHandler(channel = "spring-commands.handle")
    public CommandResponseMessage onCommand(CommandMessageReceived cmd) {
        byte[] body = cmd.getBody();
        String text = body != null ? new String(body, StandardCharsets.UTF_8) : "<empty>";
        log.info("Command handler received: channel={} body={}", cmd.getChannel(), text);
        return CommandResponseMessage.builder()
                .commandReceived(cmd)
                .isExecuted(true)
                .build();
    }
}

Both styles produce the same end-to-end result for the sender:

INFO  Command handler received: channel=spring-commands.handle body=Deploy v2.0
INFO  Command response: executed=true

Sending Queries

Queries mirror commands but return a data payload. Call sendQuery(channel, data, Duration timeout); the returned QueryResponseMessage carries both isExecuted() and a getBody() you can decode. Always check isExecuted() before reading the body, and inspect getError() when it is false.

SendQueryRunner.java

package io.kubemq.spring.boot.examples.queries;

import io.kubemq.sdk.cq.QueryResponseMessage;
import io.kubemq.spring.boot.autoconfigure.template.KubeMQTemplate;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

@Component
public class SendQueryRunner implements ApplicationRunner {

    private static final Logger log = LoggerFactory.getLogger(SendQueryRunner.class);

    private final KubeMQTemplate template;

    public SendQueryRunner(KubeMQTemplate template) {
        this.template = template;
    }

    @Override
    public void run(ApplicationArguments args) {
        QueryResponseMessage response = template.sendQuery(
                "spring-queries.send", "Get user data", Duration.ofSeconds(30));
        if (!response.isExecuted()) {
            log.warn("Query was not executed: {}", response.getError());
            return;
        }
        byte[] rawBody = response.getBody();
        String body = rawBody != null ? new String(rawBody, StandardCharsets.UTF_8) : "<empty>";
        log.info("Query response: executed={} body={}", response.isExecuted(), body);
    }
}

As with commands, sendQueryAsync(channel, data, timeout) returns a CompletableFuture<QueryResponseMessage> for non-blocking calls.

Handling Queries

Annotate a bean method with @KubeMQQueryHandler(channel = ...). The method receives a QueryMessageReceived and must return a QueryResponseMessage — unlike commands, there is no boolean shorthand, because a query response always carries a body. Build the response with .queryReceived(query) to correlate it with the request, .body(...) for the payload bytes, and .isExecuted(true).

HandleQueryHandler.java

package io.kubemq.spring.boot.examples.queries;

import io.kubemq.sdk.cq.QueryMessageReceived;
import io.kubemq.sdk.cq.QueryResponseMessage;
import io.kubemq.spring.boot.autoconfigure.listener.KubeMQQueryHandler;
import java.nio.charset.StandardCharsets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

@Component
public class HandleQueryHandler {

    private static final Logger log = LoggerFactory.getLogger(HandleQueryHandler.class);

    @KubeMQQueryHandler(channel = "spring-queries.handle")
    public QueryResponseMessage onQuery(QueryMessageReceived query) {
        String body = query.getBody() != null
                ? new String(query.getBody(), StandardCharsets.UTF_8) : "<empty>";
        log.info("Query handler received: channel={} body={}", query.getChannel(), body);
        return QueryResponseMessage.builder()
                .queryReceived(query)
                .isExecuted(true)
                .body("{\"orderId\": \"ORD-123\", \"status\": \"shipped\"}".getBytes(StandardCharsets.UTF_8))
                .build();
    }
}

The sender decodes the returned body:

INFO  Query handler received: channel=spring-queries.handle body=Get order status
INFO  Query response: executed=true body={"orderId": "ORD-123", "status": "shipped"}

Timeouts

CQ has two distinct timeouts — one on each side of the request.

Send-side timeout is the Duration you pass to sendCommand / sendQuery. The template converts it to whole seconds via durationToSeconds, rounding up to at least one second; a zero or null Duration maps to 0 (no explicit timeout). If the handler does not reply within this window, the call fails with a timeout error.

Handler-side timeout is configured per pattern and defaults to 10 seconds for both commands and queries. It bounds how long the listener container waits for your handler method to produce a response before giving up.

application.yml

kubemq:
  address: ${KUBEMQ_ADDRESS:localhost:50000}
  client-id: spring-commands-send
  listener:
    commands:
      timeout: 10s   # default — max time a command handler may take
    queries:
      timeout: 10s   # default — max time a query handler may take

When the send-side timeout is shorter than the time a handler takes, the caller times out. The example below sends with a 2-second timeout to a handler that sleeps 5 seconds, so the command never reports success:

SlowCommandHandler.java

@KubeMQCommandHandler(channel = "spring-commands.timeout")
public CommandResponseMessage onCommand(CommandMessageReceived cmd) {
    log.info("Slow handler started: channel={}", cmd.getChannel());
    try {
        TimeUnit.SECONDS.sleep(5); // exceeds the caller's 2s timeout
    } catch (InterruptedException ex) {
        Thread.currentThread().interrupt();
        return CommandResponseMessage.builder()
                .commandReceived(cmd).isExecuted(false).build();
    }
    return CommandResponseMessage.builder()
            .commandReceived(cmd).isExecuted(true).build();
}

CommandTimeoutRunner.java

try {
    CommandResponseMessage response = template.sendCommand(
            "spring-commands.timeout", "Will timeout", Duration.ofSeconds(2));
    log.info("Command response: executed={}", response.isExecuted());
} catch (Exception ex) {
    String msg = ex.getMessage() != null ? ex.getMessage().toLowerCase() : "";
    if (msg.contains("timeout") || msg.contains("deadline")) {
        log.info("Command timed out as expected: {}", ex.getMessage());
    }
}

Set the send-side timeout comfortably larger than the slowest expected handler execution. A timeout that is too tight surfaces as a failed request even when the handler eventually succeeds — work that may already be in progress on the handler side.

Consumer Groups

By default a CQ request goes to a single handler. When you register multiple handlers on the same channel with the same group, KubeMQ load-balances requests across the group members — each request is delivered to exactly one member. This lets you scale handler throughput horizontally.

Define two beans on the same channel and group (they are identical except for a log prefix):

CommandGroupHandlerA.java

@Component
public class CommandGroupHandlerA {

    private static final Logger log = LoggerFactory.getLogger(CommandGroupHandlerA.class);

    @KubeMQCommandHandler(channel = "spring-commands.consumer-group", group = "cmd-group")
    public boolean onCommand(CommandMessageReceived cmd) {
        String body = cmd.getBody() != null
                ? new String(cmd.getBody(), StandardCharsets.UTF_8) : "<empty>";
        log.info("[Handler-A] Command received: {}", body);
        return true;
    }
}

CommandGroupHandlerB.java

@Component
public class CommandGroupHandlerB {

    private static final Logger log = LoggerFactory.getLogger(CommandGroupHandlerB.class);

    @KubeMQCommandHandler(channel = "spring-commands.consumer-group", group = "cmd-group")
    public boolean onCommand(CommandMessageReceived cmd) {
        String body = cmd.getBody() != null
                ? new String(cmd.getBody(), StandardCharsets.UTF_8) : "<empty>";
        log.info("[Handler-B] Command received: {}", body);
        return true;
    }
}

Sending four commands distributes them across Handler-A and Handler-B:

CommandConsumerGroupRunner.java

for (int i = 1; i <= 4; i++) {
    CommandResponseMessage response = template.sendCommand(
            "spring-commands.consumer-group", "Command #" + i, Duration.ofSeconds(30));
    log.info("Command #{} response: executed={}", i, response.isExecuted());
}

Query handlers work the same way — register two @KubeMQQueryHandler beans with a shared group and KubeMQ balances incoming queries across them:

QueryGroupHandlerA.java

@KubeMQQueryHandler(channel = "spring-queries.consumer-group", group = "query-group")
public QueryResponseMessage onQuery(QueryMessageReceived query) {
    String body = query.getBody() != null
            ? new String(query.getBody(), StandardCharsets.UTF_8) : "<empty>";
    log.info("[Handler-A] Query received: {}", body);
    return QueryResponseMessage.builder()
            .queryReceived(query)
            .isExecuted(true)
            .body("[A] processed".getBytes(StandardCharsets.UTF_8))
            .build();
}

Cached Queries

For queries whose result is stable for a period, KubeMQ can cache the handler's response by key. The first request invokes the handler; subsequent requests with the same cache key are served from the cache until the TTL expires — the handler is not invoked again. Use the fluent query builder to set withCacheKey(...) and withCacheTTL(...):

CachedQueryRunner.java

// First call — invokes the handler and populates the cache.
QueryResponseMessage response = template.newQuery("Get config")
        .toChannel("spring-queries.cached")
        .withCacheKey("config-cache")
        .withCacheTTL(Duration.ofMinutes(5))
        .withTimeout(Duration.ofSeconds(30))
        .send();

// Second call — same cache key, served from cache without hitting the handler.
QueryResponseMessage cached = template.newQuery("Get config")
        .toChannel("spring-queries.cached")
        .withCacheKey("config-cache")
        .withCacheTTL(Duration.ofMinutes(5))
        .withTimeout(Duration.ofSeconds(30))
        .send();

The handler runs only on the first call, yet both responses carry the same body:

INFO  Cache handler invoked (first call only): body=Get config
INFO  Cached query response: executed=true body={"maxRetries": 3, "timeout": 30}
INFO  Second query (from cache): executed=true body={"maxRetries": 3, "timeout": 30}

Fluent Builders

Beyond the sendCommand / sendQuery shorthands, KubeMQTemplate exposes fluent builders for richer requests. newCommand(data) returns a KubeMQCommandMessageBuilder and newQuery(data) returns a KubeMQQueryMessageBuilder. Both let you set the channel, timeout, tags, and metadata before calling .send() (or .sendAsync() for a CompletableFuture); the query builder adds withCacheKey / withCacheTTL.

Fluent command and query

// Command with tags and a timeout.
CommandResponseMessage cmdResponse = template.newCommand("Restart service")
        .toChannel("spring-commands.send")
        .withTag("priority", "high")
        .withTimeout(Duration.ofSeconds(30))
        .send();

// Query with metadata and a timeout.
QueryResponseMessage queryResponse = template.newQuery("Get user data")
        .toChannel("spring-queries.send")
        .withMetadata("v2")
        .withTimeout(Duration.ofSeconds(30))
        .send();

The builder's withTimeout(Duration) is the send-side timeout — equivalent to the Duration argument of sendCommand / sendQuery. Cached queries are only available through the query builder, since caching needs an explicit cache key.

Kotlin

The optional kubemq-spring-boot-starter-kotlin module adds coroutine suspend extensions that bridge the *Async methods to structured concurrency. sendCommandSuspend and sendQuerySuspend suspend until the underlying CompletableFuture completes, propagating cancellation.

CommandService.java

import io.kubemq.sdk.cq.CommandResponseMessage;
import io.kubemq.sdk.cq.QueryResponseMessage;
import io.kubemq.spring.boot.autoconfigure.template.KubeMQTemplate;
import java.time.Duration;
import org.springframework.stereotype.Service;

@Service
public class CommandService {

    private final KubeMQTemplate template;

    public CommandService(KubeMQTemplate template) {
        this.template = template;
    }

    public boolean restart() {
        CommandResponseMessage response = template.sendCommand(
                "spring-commands.send", "Restart service", Duration.ofSeconds(30));
        return response.isExecuted();
    }

    public QueryResponseMessage lookupUser() {
        return template.sendQuery(
                "spring-queries.send", "Get user data", Duration.ofSeconds(30));
    }
}

CommandService.kt

import io.kubemq.sdk.cq.CommandResponseMessage
import io.kubemq.sdk.cq.QueryResponseMessage
import io.kubemq.spring.boot.autoconfigure.template.KubeMQTemplate
import io.kubemq.spring.boot.kotlin.sendCommandSuspend
import io.kubemq.spring.boot.kotlin.sendQuerySuspend
import org.springframework.stereotype.Service
import java.time.Duration

@Service
class CommandService(private val template: KubeMQTemplate) {

    suspend fun restart(): Boolean {
        val response: CommandResponseMessage = template.sendCommandSuspend(
            "spring-commands.send", "Restart service", Duration.ofSeconds(30))
        return response.isExecuted
    }

    suspend fun lookupUser(): QueryResponseMessage =
        template.sendQuerySuspend(
            "spring-queries.send", "Get user data", Duration.ofSeconds(30))
}

Commands & Queries

Sending Commands

Handling Commands

Simple mode — return a boolean

Full control — return a CommandResponseMessage

Sending Queries

Handling Queries

Timeouts

Consumer Groups

Cached Queries

Fluent Builders

Kotlin

Next Steps

Getting Started

Events & Events Store

Queues

Reference

On this page