Spring Boot Concepts

The KubeMQ Spring Boot Starter brings the broker into the Spring programming model without exposing the underlying gRPC plumbing. It does this through three layers that mirror how Spring developers already think about messaging: auto-configuration wires the SDK clients from your kubemq.* properties, the KubeMQTemplate is the injectable send-side facade, and annotation-driven listeners turn ordinary bean methods into message consumers. This page explains each layer, how the five messaging patterns map onto Spring constructs, and how the project's modules fit together.

Everything here is built on the native KubeMQ Java SDK clients — PubSubClient, QueuesClient, and CQClient — so your application speaks gRPC to the broker on port 50000 directly. The integration adds Spring lifecycle, dependency injection, configuration binding, health, and observability on top.

Auto-Configuration Model

The starter follows the standard Spring Boot auto-configuration mechanism. Six @AutoConfiguration classes are registered through the META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports file, so they activate automatically the moment the starter is on the classpath — no @Enable annotation or component scan is required.

io.kubemq.spring.boot.autoconfigure.KubeMQAutoConfiguration
io.kubemq.spring.boot.autoconfigure.KubeMQTemplateAutoConfiguration
io.kubemq.spring.boot.autoconfigure.KubeMQListenerAutoConfiguration
io.kubemq.spring.boot.autoconfigure.KubeMQHealthContributorAutoConfiguration
io.kubemq.spring.boot.autoconfigure.KubeMQMetricsAutoConfiguration
io.kubemq.spring.boot.autoconfigure.KubeMQObservationAutoConfiguration

Each class owns one responsibility:

A single master switch turns the whole integration on or off. Setting kubemq.enabled=false disables auto-configuration entirely — useful for profiles or tests where you want the application to start without connecting to a broker.

kubemq:
  enabled: true          # master switch (default: true)
  address: localhost:50000
  client-id: my-app

Configuration Properties

All settings are bound under the kubemq.* prefix by the @ConfigurationProperties-annotated KubeMQProperties class. The top level holds the connection essentials, and nested objects group the rest by concern.

@ConfigurationProperties(prefix = "kubemq")
public class KubeMQProperties {
    private boolean enabled = true;
    private String address = "localhost:50000";
    private String clientId = "";
    private String authToken = "";

    private final Tls tls = new Tls();
    private final Connection connection = new Connection();
    private final Listener listener = new Listener();
    private final Template template = new Template();
    private final Health health = new Health();
    private final Metrics metrics = new Metrics();
    private final Kotlin kotlin = new Kotlin();
    // ...
}

The nested groups map to focused areas of the integration:

The full property tables — with every key, type, and default — live in the Configuration Reference.

KubeMQTemplate

KubeMQTemplate is the thread-safe send-side facade. It wraps all three SDK clients behind a single injectable bean, applies payload serialization through a KubeMQMessageConverter, and optionally records a Micrometer Observation around each send. One template covers all five messaging patterns.

@ThreadSafe
public class KubeMQTemplate {
    private final PubSubClient pubSubClient;
    private final QueuesClient queuesClient;
    private final CQClient cqClient;
    private volatile KubeMQMessageConverter messageConverter;
    // observation registry + convention ...
}

Each pattern is exposed in three forms, so you can pick the calling style that fits the call site:

• Synchronous — blocks until the broker acknowledges (or, for commands and queries, until the response arrives): sendEvent, sendEventStore, sendQueueMessage, sendCommand, sendQuery.
• Asynchronous — returns a CompletableFuture so the caller can compose without blocking: sendEventAsync, sendEventStoreAsync, sendQueueMessageAsync, sendCommandAsync, sendQueryAsync.
• Fluent builders — a chained API for setting channel, tags, and pattern-specific options before calling .send(): newEvent, newEventStore, newQueueMessage, newCommand, newQuery.

The simplest form takes a channel and a payload object. The converter serializes the object to bytes (falling back to handling byte[] and String directly when no converter is configured).

@Service
public class OrderService {
    private final KubeMQTemplate template;

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

    // Synchronous fire-and-forget event
    public void placeOrder(Order order) {
        template.sendEvent("orders", order);
    }

    // Asynchronous send returning a CompletableFuture
    public CompletableFuture<Void> placeOrderAsync(Order order) {
        return template.sendEventAsync("orders", order);
    }

    // Synchronous request-response query
    public QueryResponseMessage lookup(String key) {
        return template.sendQuery("queries.lookup", key, Duration.ofSeconds(10));
    }
}

The fluent builders shine when a send needs more than a channel and a body — tags, delays, expirations, timeouts, or query cache settings:

template.newEvent("Tagged event")
        .toChannel("spring-fluent.events")
        .withTag("source", "fluent-builder")
        .send();

template.newQueueMessage("Delayed queue msg")
        .toChannel("spring-fluent.queues")
        .withDelay(Duration.ofSeconds(1))
        .withExpiration(Duration.ofSeconds(30))
        .send();

CommandResponseMessage cmdResponse = template.newCommand("Fluent command")
        .toChannel("spring-fluent.commands")
        .withTimeout(Duration.ofSeconds(10))
        .send();

QueryResponseMessage queryResponse = template.newQuery("Fluent query")
        .toChannel("spring-fluent.queries")
        .withTimeout(Duration.ofSeconds(10))
        .withCacheKey("fluent-cache")
        .withCacheTTL(Duration.ofMinutes(5))
        .send();

Annotation-Driven Listeners

The receive side mirrors Spring's familiar listener model. You annotate a bean method, and the integration subscribes to the broker and dispatches each message to that method. Discovery is performed by KubeMQListenerAnnotationBeanPostProcessor, which scans every Spring bean for the five listener annotations and registers a container for each one it finds.

ReflectionUtils.doWithMethods(targetClass, method -> {
    processAnnotation(bean, method, KubeMQEventListener.class, KubeMQListenerType.EVENT);
    processAnnotation(bean, method, KubeMQEventStoreListener.class, KubeMQListenerType.EVENT_STORE);
    processAnnotation(bean, method, KubeMQQueueListener.class, KubeMQListenerType.QUEUE);
    processAnnotation(bean, method, KubeMQCommandHandler.class, KubeMQListenerType.COMMAND);
    processAnnotation(bean, method, KubeMQQueryHandler.class, KubeMQListenerType.QUERY);
});

Five annotations map to the five messaging patterns:

The two command/query annotations use a singular channel attribute because CQ handlers are point-to-point: each handler services exactly one channel. The three pub/sub and queue annotations use a plural channels array so one method can serve several channels.

@Component
public class OrderConsumer {

    @KubeMQEventListener(channels = "orders", group = "order-group")
    public void onOrder(EventMessageReceived event) {
        // process event
    }

    @KubeMQQueueListener(channels = "queues.orders", autoAck = "true")
    public void processOrder(QueueMessageReceived msg) {
        // process queue message
    }

    @KubeMQQueryHandler(channel = "queries.user-lookup")
    public QueryResponseMessage handleQuery(QueryMessageReceived query) {
        return QueryResponseMessage.builder()
                .queryReceived(query)
                .body(lookupResult)
                .isExecuted(true)
                .build();
    }
}

Notice that every annotation attribute is a String, even numeric and boolean ones such as concurrency, autoAck, and autoStartup. This is deliberate: each attribute is resolved through Environment.resolvePlaceholders at startup, so all attributes support SpEL expressions and property placeholders. That lets you externalize channel names and tuning into configuration:

@KubeMQEventListener(
    channels = "${kubemq.channels.orders}",
    group = "order-group",
    concurrency = "${app.listener.concurrency:4}"
)
public void onOrder(EventMessageReceived event) { ... }

The post-processor also validates each handler at startup. Query handlers must return QueryResponseMessage; command handlers may return void, boolean/Boolean, or CommandResponseMessage; and each method must take exactly one parameter of the expected received-message type (a Kotlin Continuation is permitted as a trailing parameter for suspend functions). A mismatch fails fast with a BeanCreationException rather than at first message.

Listener Containers and Concurrency

Each discovered endpoint is materialized as a KubeMQMessageListenerContainer, produced by the KubeMQListenerContainerFactory. The container implements Spring's SmartLifecycle, so it starts after all singletons are initialized and stops cleanly during application shutdown — the SDK handles broker reconnection transparently while the container stays alive.

Container behaviour is driven by kubemq.listener.* (with per-listener annotation attributes overriding the defaults):

• concurrency — number of concurrent message processors (or poll loops for queues). Defaults to 1.
• auto-startup — whether containers start with the application context. Defaults to true.
• shutdown-timeout — how long to wait for in-flight work to drain on stop. Defaults to 30s.

The fundamental difference between patterns is reflected in how the container consumes messages:

• Events and Events Store are push-based. The container opens a subscription and the broker pushes messages to a callback, bounded by maxConcurrentCallbacks. There are no poll settings.
• Queues are poll-based. The container runs a poll loop per channel, requesting batches with a poll timeout, visibility timeout, and optional auto-ack. On poll errors it applies exponential backoff (initial → max, by a multiplier) before retrying.
• Commands and Queries are request-response subscriptions. The container subscribes, invokes the handler, and sends the handler's return value back to the broker as the response.

QueuesPollRequest request = QueuesPollRequest.builder()
        .channel(channel)
        .pollMaxMessages(maxMessages)
        .pollWaitTimeoutInSeconds(pollTimeoutSeconds)
        .autoAckMessages(endpoint.isAutoAck())
        .visibilitySeconds(visibilitySeconds)
        .build();
QueuesPollResponse response = queuesClient.receiveQueueMessages(request);

Messaging Patterns and Spring Constructs

The integration deliberately splits the five patterns into two annotation families that match their semantics — listeners for one-way delivery and handlers for request-response.

The events-store annotation adds replay control through subscriptionType (StartNewOnly, StartFromFirst, StartFromLast, StartAtSequence, StartAtTime, StartAtTimeDelta) and a matching subscriptionValue. The command and query handlers are the only patterns whose return value is sent back to the broker — that is what makes them request-response. For the full send-and-receive walkthroughs, see the pattern pages linked in Related Topics.

Send and Receive Paths

The two diagrams below show how a message travels through the integration. On the send path, application code calls the template, which converts the payload and delegates to the matching SDK client. On the receive path, the SDK subscription (push) or poll loop (queues) delivers a message to the listener container, which dispatches it to your annotated method.

Send path: a bean calls the template, which serializes through the converter and delegates to the matching SDK client over gRPC.

Receive path: the SDK subscription or poll loop feeds the listener container, which dispatches to your annotated method (and returns a response for commands/queries).

Module Map

The project is a multi-module Gradle build. You add kubemq-spring-boot-starter to your application; the others are pulled in transitively or added when you need Spring Cloud Stream, Kotlin, or testing support.

The test module's @KubeMQTest annotation composes with @SpringBootTest and selects an infrastructure mode — MOCK (in-process gRPC, no Docker), EMBEDDED (TestContainers), or EXTERNAL (an existing broker):

@KubeMQTest(mode = KubeMQTestMode.MOCK)
class MyEventTest {

    @Autowired
    KubeMQTemplate template;

    @Autowired
    MockKubeMQServer mockServer;

    @Test
    void shouldSendEvent() {
        template.sendEvent("test-channel", "hello");
        // assert against mockServer ...
    }
}

Observability Concepts

The integration integrates with Micrometer for both tracing-style observations and health, all toggleable through kubemq.* properties.

• Observations. When an ObservationRegistry is present, the template records a kubemq.send observation around every send (KubeMQSendObservation), and listener containers record a kubemq.receive observation around every delivery (KubeMQReceiveObservation). Both carry a low-cardinality kubemq.pattern tag (EVENTS, EVENTS_STORE, QUEUES, COMMANDS, QUERIES) and a high-cardinality kubemq.channel tag. Send observations are gated by kubemq.template.observation-enabled.
• Health. KubeMQHealthIndicator pings all three clients and reports UP with the broker host and version when all respond, or DOWN with the failing exception type. Results are cached for kubemq.health.cache-duration (default 15s) so health probes don't overwhelm the broker, and each ping is bounded by kubemq.health.timeout (default 5s).
• Actuator endpoint. KubeMQHealthContributorAutoConfiguration also registers a dedicated /actuator/kubemq endpoint. It probes the pubsub, queues, and cq clients independently and returns a per-client status of connected or disconnected, with the broker host and version for each. The overall status is connected only when all three respond, otherwise degraded.

{
  "status": "connected",
  "clients": {
    "pubsub": { "status": "connected", "host": "localhost", "version": "...", "errorType": null },
    "queues": { "status": "connected", "host": "localhost", "version": "...", "errorType": null },
    "cq":     { "status": "connected", "host": "localhost", "version": "...", "errorType": null }
  }
}

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

Related Topics