Overview

Request-reply gives you synchronous RPC on top of KubeMQ's messaging fabric: a caller sends a query and blocks until the handler actually processing the request sends back a real answer — not just an acknowledgment. Reach for it whenever the caller needs a return value to proceed — a lookup, a computed result, a status check — the same shape as an HTTP call, but routed by KubeMQ instead of a service mesh or DNS.

A handler subscribes with subscribeToQueries and, inside its onReceiveQueryCallback, builds a QueryResponseMessage via .queryReceived(request), which copies the request's correlation ID so KubeMQ can route the response to the one caller waiting, not broadcast it. The caller's client.sendQuery(...) blocks the calling thread until that reply lands or timeoutInSeconds elapses, throwing a KubeMQTimeoutException on timeout.

Gotchas: if no subscriber is listening — or the handler crashes before replying — sendQuery simply times out; there's no way to distinguish "no handler" from "handler is slow" from the exception alone. QueryResponseMessage.builder().queryReceived(request) must reference the exact request object it's answering — build a fresh one and the correlation ID is lost, so the reply is silently dropped or misrouted. If you don't actually need a return value, use commands instead — they only need an ack, so they don't tie up a caller waiting on a round trip.

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

RequestReplyExample.java

package io.kubemq.example.patterns;

import io.kubemq.sdk.cq.*;
import io.kubemq.sdk.common.ServerInfo;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;

/**
 * Request-Reply Pattern Example
 *
 * Demonstrates implementing the Request-Reply pattern using KubeMQ Queries.
 */
public class RequestReplyExample {

    private static final String ADDRESS = "localhost:50000";
    private static final String CLIENT_ID = "java-patterns-request-reply-client";
    private static final String CHANNEL = "java-patterns.request-reply";

    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();
        ServerInfo info = client.ping();
        System.out.println("Connected to: " + info.getHost());
        // Create the queries channel for request-reply
        client.createQueriesChannel(CHANNEL);

        // Define the service handler (receives request, sends response)
        Consumer<QueryMessageReceived> service = request -> {
            String requestBody = new String(request.getBody());
            System.out.println("  [Service] Received: " + requestBody);

            String response = "{\"userId\": 123, \"name\": \"John Doe\"}";
            Map<String, String> tags = new HashMap<>();
            tags.put("status", "200");

            client.sendResponseMessage(QueryResponseMessage.builder()
                    .queryReceived(request).isExecuted(true)
                    .body(response.getBytes(StandardCharsets.UTF_8)).tags(tags).build());
        };

        QueriesSubscription sub = QueriesSubscription.builder()
                .channel(CHANNEL).onReceiveQueryCallback(service)
                .onErrorCallback(err -> System.err.println("Error: " + err)).build();

        // Subscribe to handle incoming requests
        client.subscribeToQueries(sub);
        Thread.sleep(300);

        // Client sends request and waits for reply
        System.out.println("\nClient sending request...");
        QueryResponseMessage response = client.sendQuery(QueryMessage.builder()
                .channel(CHANNEL).body("getUser:123".getBytes()).timeoutInSeconds(10).build());
        System.out.println("Response: " + new String(response.getBody()));
        System.out.println("Status: " + response.getTags().get("status"));

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

How It Works

The service handler subscribes via QueriesSubscription with an onReceiveQueryCallback; inside the callback it calls client.sendResponseMessage(QueryResponseMessage.builder().queryReceived(request)...) to route the response back to the original caller.
QueryResponseMessage.builder().queryReceived(request) copies the request's correlation ID — KubeMQ uses this ID to match the response to the blocked sendQuery() caller.
sendQuery() blocks the calling thread until the response arrives or timeoutInSeconds elapses; a KubeMQTimeoutException is thrown on timeout.
Both the sender and the service share one CQClient instance here for brevity; in production each component would have its own client and likely run in separate processes.

Request-Reply

Overview

Prerequisites

Code

How It Works

Related

On this page