# Spring Boot Concepts (/integrations/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 [#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. ```text title="META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports" 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: | Auto-configuration class | Responsibility | | ------------------------------------------ | ----------------------------------------------------------------------------------------------------- | | `KubeMQAutoConfiguration` | Builds the three SDK clients (`PubSubClient`, `QueuesClient`, `CQClient`) from `kubemq.*` properties. | | `KubeMQTemplateAutoConfiguration` | Exposes the `KubeMQTemplate` send-side bean. | | `KubeMQListenerAutoConfiguration` | Registers the listener bean post-processor, endpoint registrar, and container factory. | | `KubeMQHealthContributorAutoConfiguration` | Contributes the `KubeMQHealthIndicator` and the `/actuator/kubemq` endpoint. | | `KubeMQMetricsAutoConfiguration` | Wires Micrometer metrics binding. | | `KubeMQObservationAutoConfiguration` | Wires the Micrometer Observation conventions for send and receive. | 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. ```yaml title="application.yml" kubemq: enabled: true # master switch (default: true) address: localhost:50000 client-id: my-app ``` Because the auto-configurations are conditional, you can override any bean by declaring your own. For example, defining your own `KubeMQTemplate` or `KubeMQMessageConverter` bean replaces the auto-configured one. ## Configuration Properties [#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. ```java title="KubeMQProperties.java" @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: | Property group | Prefix | Covers | | -------------- | -------------------- | --------------------------------------------------------------------------------------------------------- | | Top level | `kubemq.` | `enabled`, `address`, `client-id`, `auth-token` | | TLS / mTLS | `kubemq.tls.` | `enabled`, `cert-file`, `key-file`, `ca-cert-file` | | Connection | `kubemq.connection.` | `timeout`, `max-receive-size`, `keep-alive.*` | | Listener | `kubemq.listener.` | `concurrency`, `auto-startup`, `shutdown-timeout`, plus per-pattern `queues.*`, `commands.*`, `queries.*` | | Template | `kubemq.template.` | `observation-enabled` | | Health | `kubemq.health.` | `enabled`, `timeout`, `cache-duration` | | Metrics | `kubemq.metrics.` | `enabled`, `scrape-interval` | | Kotlin | `kubemq.kotlin.` | `dispatcher` (`default`, `io`, or `unconfined`) | The `client-id` defaults to an empty string — when left unset, the SDK generates a UUID per client. Duration values such as `kubemq.connection.timeout` and `kubemq.listener.shutdown-timeout` accept Spring's relaxed duration syntax (for example `30s` or `5m`), and `kubemq.connection.max-receive-size` accepts data-size syntax (for example `100MB`). The full property tables — with every key, type, and default — live in the [Configuration Reference](/integrations/spring-boot/reference/configuration). ## KubeMQTemplate [#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. ```java title="KubeMQTemplate.java" @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). ```java title="OrderService.java" @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 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: ```java title="FluentBuildersRunner.java" 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(); ``` When `kubemq.template.observation-enabled` is `true` (the default) and an `ObservationRegistry` is present, every send is wrapped in a `kubemq.send` observation tagged with the channel and pattern. See [Observability concepts](#observability-concepts) below. ## Annotation-Driven Listeners [#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. ```java title="KubeMQListenerAnnotationBeanPostProcessor.java" 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: | Annotation | Pattern | Method parameter | Channel attribute | | --------------------------- | ------------ | ------------------------------------------------------ | --------------------- | | `@KubeMQEventListener` | Events | `EventMessageReceived` | `channels` (multiple) | | `@KubeMQEventStoreListener` | Events Store | `EventStoreMessageReceived` | `channels` (multiple) | | `@KubeMQQueueListener` | Queues | `QueueMessageReceived` or `List` | `channels` (multiple) | | `@KubeMQCommandHandler` | Commands | `CommandMessageReceived` | `channel` (single) | | `@KubeMQQueryHandler` | Queries | `QueryMessageReceived` | `channel` (single) | 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. ```java title="OrderConsumer.java" @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: ```java title="ConfigurableListener.java" @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 [#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. ```java title="QueuesPollRequest (built inside the container)" QueuesPollRequest request = QueuesPollRequest.builder() .channel(channel) .pollMaxMessages(maxMessages) .pollWaitTimeoutInSeconds(pollTimeoutSeconds) .autoAckMessages(endpoint.isAutoAck()) .visibilitySeconds(visibilitySeconds) .build(); QueuesPollResponse response = queuesClient.receiveQueueMessages(request); ``` With `autoAck = "false"` (the default for queues), messages remain invisible for `visibilityTimeout` seconds and must be acknowledged explicitly. If the handler does not ack within that window, the broker redelivers the message. Setting `autoAck = "true"` acknowledges on receipt, trading at-least-once redelivery for simpler handling. ## Messaging Patterns and Spring Constructs [#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. | Pattern | Spring construct | Delivery | Return value | | ------------ | --------------------------- | -------------------------------------------------- | --------------------------------------------- | | Events | `@KubeMQEventListener` | Push subscription, fire-and-forget | Ignored | | Events Store | `@KubeMQEventStoreListener` | Push subscription with replay (`subscriptionType`) | Ignored | | Queues | `@KubeMQQueueListener` | Poll loop, durable point-to-point | Ignored; ack/visibility governs redelivery | | Commands | `@KubeMQCommandHandler` | Request-response subscription | `void` / `boolean` / `CommandResponseMessage` | | Queries | `@KubeMQQueryHandler` | Request-response subscription | `QueryResponseMessage` | 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](#related-topics). ## Send and Receive Paths [#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 [#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. | Module | Role | | ----------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------- | | `kubemq-spring-boot-autoconfigure` | Core beans — the six auto-configurations, `KubeMQTemplate`, listener annotations and containers, health, metrics, and observation. | | `kubemq-spring-boot-starter` | Dependency aggregator — the single artifact you add to a project; pulls in the autoconfigure module and the KubeMQ Java SDK. | | `kubemq-spring-cloud-stream-binder` | A Spring Cloud Stream binder for Events, Events Store, and Queues, exposing KubeMQ through the binder programming model. | | `kubemq-spring-boot-starter-kotlin` | Kotlin support — coroutine extensions, `Flow` adapters for subscriptions, and a configuration DSL. | | `kubemq-spring-boot-starter-test` | Test harness — `MockKubeMQServer` (in-process gRPC mock), TestContainers support, and the `@KubeMQTest` annotation. | 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): ```java title="MyEventTest.java" @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 [#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`. ```json title="GET /actuator/kubemq" { "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 } } } ``` A local broker exposes the gRPC API the SDK clients connect to on `50000` and the shared HTTP/dashboard endpoints on `9090`: ## Related Topics [#related-topics]