This page is the authoritative reference for the KubeMQ Spring Boot Starter's runtime API: the KubeMQTemplate send methods and fluent builders, the listener and handler annotation attributes, the Spring Cloud Stream binder properties, and the Actuator endpoints. For module coordinates and the full kubemq.* property tables, see the configuration reference.

KubeMQTemplate API

KubeMQTemplate is a thread-safe, injectable bean that covers all five messaging patterns. It wraps the native SDK clients (PubSubClient, QueuesClient, CQClient) with payload serialization and optional Micrometer observation. Inject it through the constructor:

OrderService.java

@Service
public class OrderService {
    private final KubeMQTemplate template;

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

Send Methods

channel is the target channel, data is the payload (serialized by the configured KubeMQMessageConverter, with a built-in fallback for byte[] and String), and tags is an optional Map<String, String>. Command and query sends take a Duration timeout.

Method	Returns	Description
`sendEvent(String channel, Object data)`	`void`	Fire-and-forget event.
`sendEvent(String channel, Object data, Map<String, String> tags)`	`void`	Event with tags.
`sendEventAsync(String channel, Object data)`	`CompletableFuture<Void>`	Async event.
`sendEventAsync(String channel, Object data, Map<String, String> tags)`	`CompletableFuture<Void>`	Async event with tags.
`sendEventStore(String channel, Object data)`	`void`	Persistent, replayable event.
`sendEventStore(String channel, Object data, Map<String, String> tags)`	`void`	Persistent event with tags.
`sendEventStoreAsync(String channel, Object data)`	`CompletableFuture<Void>`	Async persistent event.
`sendEventStoreAsync(String channel, Object data, Map<String, String> tags)`	`CompletableFuture<Void>`	Async persistent event with tags.
`sendQueueMessage(String channel, Object data)`	`void`	Single durable queue message.
`sendQueueMessage(String channel, Object data, Map<String, String> tags)`	`void`	Queue message with tags.
`sendQueueMessages(String channel, List<?> data)`	`void`	Multiple queue messages to one channel. Empty/`null` lists are a no-op.
`sendQueueMessageAsync(String channel, Object data)`	`CompletableFuture<Void>`	Async queue message.
`sendQueueMessageAsync(String channel, Object data, Map<String, String> tags)`	`CompletableFuture<Void>`	Async queue message with tags.
`sendCommand(String channel, Object data, Duration timeout)`	`CommandResponseMessage`	RPC command; blocks until the handler responds or the timeout elapses.
`sendCommandAsync(String channel, Object data, Duration timeout)`	`CompletableFuture<CommandResponseMessage>`	Async RPC command.
`sendQuery(String channel, Object data, Duration timeout)`	`QueryResponseMessage`	RPC query; blocks and returns the response body.
`sendQueryAsync(String channel, Object data, Duration timeout)`	`CompletableFuture<QueryResponseMessage>`	Async RPC query.

SendExamples.java

// Events
template.sendEvent("orders", order);
template.sendEvent("orders", order, Map.of("region", "eu"));
template.sendEventAsync("orders", order);

// Events Store (persistent, replayable)
template.sendEventStore("orders.audit", auditEntry);

// Queues
template.sendQueueMessage("queues.orders", order);
template.sendQueueMessages("queues.orders", List.of(order1, order2, order3));

// Commands & Queries (request-response)
CommandResponseMessage cmd =
    template.sendCommand("commands.device", payload, Duration.ofSeconds(10));

QueryResponseMessage qry =
    template.sendQuery("queries.user-lookup", request, Duration.ofSeconds(10));

Fluent Builders

Each fluent factory returns a builder; chain toChannel(...), optional withTag / withTags / withMetadata, plus pattern-specific options, then call the terminal send() (or sendAsync()).

Factory	Returns	Pattern-specific options	Terminal ops
`newEvent(Object data)`	`KubeMQEventMessageBuilder`	—	`send()` → `void`, `sendAsync()` → `CompletableFuture<Void>`
`newEventStore(Object data)`	`KubeMQEventMessageBuilder`	—	`send()` → `void`, `sendAsync()` → `CompletableFuture<Void>`
`newQueueMessage(Object data)`	`KubeMQQueueMessageBuilder`	`withDelay(Duration)`, `withExpiration(Duration)`, `withDeadLetterQueue(String, int)`	`send()` → `void`, `sendAsync()` → `CompletableFuture<Void>`
`newCommand(Object data)`	`KubeMQCommandMessageBuilder`	`withTimeout(Duration)`	`send()` → `CommandResponseMessage`, `sendAsync()` → `CompletableFuture<CommandResponseMessage>`
`newQuery(Object data)`	`KubeMQQueryMessageBuilder`	`withTimeout(Duration)`, `withCacheKey(String)`, `withCacheTTL(Duration)`	`send()` → `QueryResponseMessage`, `sendAsync()` → `CompletableFuture<QueryResponseMessage>`

FluentBuilders.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();

Configure a KubeMQMessageConverter bean (for example a Jackson-based JSON converter) to serialize arbitrary payload types. Without a converter, KubeMQTemplate only accepts byte[] and String payloads and throws IllegalArgumentException for anything else.

Listener Annotations

Listener and handler annotations turn bean methods into message consumers. Attribute values are String because they resolve SpEL expressions and ${...} property placeholders at runtime; an empty default means "fall back to the kubemq.listener.* configuration". Place the annotations on @Component (or other Spring bean) methods.

@KubeMQEventListener

Receives EventMessageReceived from one or more channels (push-based, fire-and-forget).

Attribute	Type	Default	Description
`channels`	`String[]`	(required)	Channel names to subscribe to. Supports SpEL and placeholders.
`group`	`String`	`""`	Consumer group for shared subscriptions. Empty means no group.
`concurrency`	`String`	`""`	Number of concurrent processors (resolved as integer).
`containerFactory`	`String`	`""`	Bean name of a custom `KubeMQListenerContainerFactory`.
`autoStartup`	`String`	`""`	Whether the listener auto-starts (resolved as boolean).
`errorHandler`	`String`	`""`	Bean name of a custom `ErrorHandler`.
`id`	`String`	`""`	Unique listener endpoint identifier.

EventListenerExample.java

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

@KubeMQEventStoreListener

Receives EventStoreMessageReceived with replay support. The subscriptionType controls the replay start position.

Attribute	Type	Default	Description
`channels`	`String[]`	(required)	Channel names to subscribe to. Supports SpEL and placeholders.
`group`	`String`	`""`	Consumer group for shared subscriptions.
`subscriptionType`	`String`	`StartNewOnly`	Replay start type (see values below).
`subscriptionValue`	`String`	`0`	Value for the subscription type, e.g. sequence number or timestamp (resolved as long).
`concurrency`	`String`	`""`	Number of concurrent processors.
`containerFactory`	`String`	`""`	Bean name of a custom `KubeMQListenerContainerFactory`.
`autoStartup`	`String`	`""`	Whether the listener auto-starts.
`errorHandler`	`String`	`""`	Bean name of a custom `ErrorHandler`.
`id`	`String`	`""`	Unique listener endpoint identifier.

subscriptionType accepts one of: StartNewOnly, StartFromFirst, StartFromLast, StartAtSequence, StartAtTime, StartAtTimeDelta.

EventStoreListenerExample.java

@KubeMQEventStoreListener(
    channels = "events_store.orders",
    subscriptionType = "StartFromFirst"
)
public void replayOrders(EventStoreMessageReceived event) {
    // replay from the first stored event
}

@KubeMQQueueListener

Polls QueueMessageReceived (single) or List<QueueMessageReceived> (batch mode) from one or more channels.

Attribute	Type	Default	Description
`channels`	`String[]`	(required)	Channel names to poll from. Supports SpEL and placeholders.
`concurrency`	`String`	`""`	Number of concurrent poll loops.
`pollTimeout`	`String`	`""`	Max seconds to wait for messages per poll cycle.
`maxPollMessages`	`String`	`""`	Max messages returned per poll.
`visibilityTimeout`	`String`	`""`	Seconds received messages stay hidden from other consumers.
`autoAck`	`String`	`""`	Whether messages are acknowledged automatically on receipt.
`batch`	`String`	`false`	Whether the method receives a `List<QueueMessageReceived>` batch.
`containerFactory`	`String`	`""`	Bean name of a custom `KubeMQListenerContainerFactory`.
`autoStartup`	`String`	`""`	Whether the listener auto-starts.
`errorHandler`	`String`	`""`	Bean name of a custom `ErrorHandler`.
`id`	`String`	`""`	Unique listener endpoint identifier.

QueueListenerExample.java

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

@KubeMQCommandHandler

Handles one command channel (point-to-point). Returns boolean (simple mode) or CommandResponseMessage (full control). Uses singular channel.

Attribute	Type	Default	Description
`channel`	`String`	(required)	The single command channel to subscribe to. Supports SpEL and placeholders.
`group`	`String`	`""`	Consumer group for shared subscriptions.
`concurrency`	`String`	`""`	Number of concurrent command processors.
`containerFactory`	`String`	`""`	Bean name of a custom `KubeMQListenerContainerFactory`.
`autoStartup`	`String`	`""`	Whether the handler auto-starts.
`errorHandler`	`String`	`""`	Bean name of a custom `ErrorHandler`.
`id`	`String`	`""`	Unique listener endpoint identifier.

CommandHandlerExample.java

@KubeMQCommandHandler(channel = "commands.device-control", group = "handlers")
public boolean handleCommand(CommandMessageReceived cmd) {
    return true; // executed successfully
}

@KubeMQQueryHandler

Handles one query channel (point-to-point). Must return a QueryResponseMessage. Uses singular channel.

Attribute	Type	Default	Description
`channel`	`String`	(required)	The single query channel to subscribe to. Supports SpEL and placeholders.
`group`	`String`	`""`	Consumer group for shared subscriptions.
`concurrency`	`String`	`""`	Number of concurrent query processors.
`containerFactory`	`String`	`""`	Bean name of a custom `KubeMQListenerContainerFactory`.
`autoStartup`	`String`	`""`	Whether the handler auto-starts.
`errorHandler`	`String`	`""`	Bean name of a custom `ErrorHandler`.
`id`	`String`	`""`	Unique listener endpoint identifier.

QueryHandlerExample.java

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

Spring Cloud Stream Binder

The kubemq-spring-cloud-stream-binder module registers a binder named kubemq (declared in META-INF/spring.binders). It binds three KubeMQ patterns, selected per binding via the pattern consumer/producer property. For the full programming-model walkthrough, see Spring Cloud Stream Binder.

KubeMQPattern values:

Value	Maps to
`EVENTS`	Fire-and-forget Events.
`EVENTS_STORE`	Persistent, replayable Events Store.
`QUEUES`	Durable point-to-point Queues.

Consumer properties are bound from KubeMQConsumerProperties under spring.cloud.stream.kubemq.bindings.<binding>.consumer.* (or spring.cloud.stream.kubemq.default.consumer.* for all bindings):

Property	Type	Default	Description
`pattern`	`KubeMQPattern`	`EVENTS`	Messaging pattern for this binding.
`eventsStoreType`	`EventsStoreType`	`StartNewOnly`	Events Store replay start type.
`eventsStoreSequenceValue`	`long`	`0`	Value for the Events Store start type (e.g. sequence number).
`pollMaxMessages`	`int`	`1`	Max messages returned per poll (Queues).
`pollWaitTimeoutInSeconds`	`int`	`5`	Poll wait timeout in seconds (Queues).
`visibilitySeconds`	`int`	`30`	Visibility timeout in seconds (Queues).
`autoAckMessages`	`boolean`	`false`	Whether queue messages are auto-acknowledged.

application.yml

spring:
  cloud:
    function:
      definition: uppercase
    stream:
      default-binder: kubemq
      bindings:
        uppercase-in-0:
          destination: spring-scs.events.in
          group: scs-events-demo
        uppercase-out-0:
          destination: spring-scs.events.out
      kubemq:
        default:
          consumer:
            pattern: EVENTS
          producer:
            pattern: EVENTS

kubemq:
  address: ${KUBEMQ_ADDRESS:localhost:50000}
  client-id: spring-scs-events

Set pattern: QUEUES (with the poll/visibility consumer properties) or pattern: EVENTS_STORE (with eventsStoreType / eventsStoreSequenceValue) to bind the other patterns.

Actuator Endpoints

When kubemq.health.enabled is true, the starter contributes a kubemq health indicator to /actuator/health. It pings all three clients (PubSubClient, QueuesClient, CQClient) and caches the result for kubemq.health.cache-duration.

Endpoint	Output
`/actuator/health`	`kubemq` component reports `UP` with `host` and `version` details when all clients respond to a ping, or `DOWN` with an `error` detail on failure.
`/actuator/kubemq`	Overall `status` (`connected` when all clients are reachable, otherwise `degraded`) plus a per-client map (`pubsub`, `queues`, `cq`).

The KubeMQEndpoint (/actuator/kubemq) reports each client as a ClientStatus record with status, host, version, and errorType:

GET /actuator/kubemq

{
  "status": "connected",
  "clients": {
    "pubsub": { "status": "connected", "host": "kubemq:50000", "version": "v3.x", "errorType": null },
    "queues": { "status": "connected", "host": "kubemq:50000", "version": "v3.x", "errorType": null },
    "cq":     { "status": "connected", "host": "kubemq:50000", "version": "v3.x", "errorType": null }
  }
}

Expose the endpoint through standard Actuator configuration: