Overview

This tutorial builds the RPC half of KubeMQ's request/reply patterns: a query, where the caller blocks for a handler's data payload instead of just a completion status. Reach for it whenever a caller needs an answer — fetching a record, running a lookup, or asking another service to compute a value on demand. You'll run a handler and a sender in the same process to see the full round trip.

The sender calls client.sendQuery() with a QueryMessage and a timeoutInSeconds, then blocks until a reply arrives. client.subscribeToQueries() registers a handler via onReceiveQueryCallback; the handler calls client.sendResponseMessage() with a QueryResponseMessage.builder().queryReceived(query) — which copies the request's correlation ID automatically — plus isExecuted(true) and a body, and KubeMQ routes that reply back to the caller waiting on it.

Gotchas: the timeout must cover however long the handler takes to run — a slow handler throws KubeMQTimeoutException even though the handler eventually succeeds. No handler subscribed yet also times out rather than erroring immediately, so startup order matters. The body is an arbitrary byte array — encoding it (StandardCharsets.UTF_8 for text) is your application's job.

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

SendQueryExample.java

package io.kubemq.example.queries;

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

public class SendQueryExample {
    private static final String ADDRESS = "localhost:50000";
    private static final String CLIENT_ID = "java-queries-send-query-client";
    private static final String CHANNEL = "java-queries.send-query";

    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 to handle incoming queries
        QueriesSubscription sub = QueriesSubscription.builder()
                .channel(CHANNEL)
                .onReceiveQueryCallback(query -> {
                    System.out.println("  Handler received: " + new String(query.getBody()));
                    client.sendResponseMessage(QueryResponseMessage.builder()
                            .queryReceived(query).isExecuted(true)
                            .body("{\"result\": \"data\"}".getBytes()).build());
                })
                .onErrorCallback(err -> System.err.println("Error: " + err))
                .build();
        // Start the query handler subscription
        client.subscribeToQueries(sub);
        Thread.sleep(300);

        Map<String, String> tags = new HashMap<>();
        tags.put("type", "lookup");

        QueryMessage query = QueryMessage.builder()
                .channel(CHANNEL).body("Get user data".getBytes())
                .metadata("Query metadata").tags(tags).timeoutInSeconds(10).build();

        // Send a query and wait for the response
        QueryResponseMessage response = client.sendQuery(query);
        System.out.println("Query executed: " + response.isExecuted());
        System.out.println("Response body: " + new String(response.getBody()));

        // Clean up resources
        sub.cancel();
        client.deleteQueriesChannel(CHANNEL);
        client.close();
    }
}

// Expected output:
//   Handler received: Get user data
// Query executed: true
// Response body: {"result": "data"}

How It Works

sendQuery() blocks the calling thread until a handler calls sendResponseMessage() with the matching correlation ID, or until timeoutInSeconds elapses with a KubeMQTimeoutException.
The inline handler subscribes and sends a response on the same CQClient instance; in production the handler typically runs in a separate service.
QueryResponseMessage.builder().queryReceived(query) copies the request's ID automatically so KubeMQ can route the response back to the blocked caller.
Response bodies are arbitrary byte arrays — the example uses a JSON string; use StandardCharsets.UTF_8 explicitly for cross-platform safety.

Send Query

Overview

Prerequisites

Code

How It Works

On this page

Send Query

Overview

Prerequisites

Code

How It Works

Related

On this page