Overview

KubeMQ provides two publish/subscribe patterns, both exposed through the Spring Boot starter:

Events — fire-and-forget, multicast delivery. An event is fanned out to every active subscriber on the channel and is not persisted. If no subscriber is connected when the event is published, it is lost.
Events Store — the same pub/sub model, but messages are persisted to the broker and assigned a monotonically increasing sequence number. New or reconnecting subscribers can replay historical events from a chosen starting point.

In the Spring adapter both patterns are push-based subscriptions: you annotate a bean method with @KubeMQEventListener or @KubeMQEventStoreListener, and the auto-configured listener container opens a streaming gRPC subscription to the broker and invokes your method for each delivered message. Publishing is done through the injectable KubeMQTemplate.

This page documents the Spring API for these two patterns. For the underlying broker semantics, see the core Events and Events Store concepts.

The template publishes events; the broker pushes to annotated listeners, and Events Store persists the log for replay.

The starter speaks gRPC to the broker on port 50000. Start a local KubeMQ broker with Docker — the gRPC API listens on 50000 and the shared HTTP/REST and dashboard endpoints on 9090:

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

Point the application at the broker in application.yml:

application.yml

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

Publishing Events

Inject KubeMQTemplate and call sendEvent(channel, data). The template serializes the payload through the configured message converter (falling back to byte[] / String when none is set) and publishes a fire-and-forget event.

BasicPubSubRunner.java

@Component
public class BasicPubSubRunner implements ApplicationRunner {

    private static final Logger log = LoggerFactory.getLogger(BasicPubSubRunner.class);

    private final KubeMQTemplate template;

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

    @Override
    public void run(ApplicationArguments args) throws InterruptedException {
        // Allow annotated listeners to subscribe before sending.
        Thread.sleep(500);
        for (int i = 1; i <= 3; i++) {
            template.sendEvent("spring-events.basic-pubsub", "Hello KubeMQ #" + i);
            log.info("Published event to spring-events.basic-pubsub: Hello KubeMQ #{}", i);
        }
    }
}

PublishRunner.kt

@Component
class PublishRunner(private val template: KubeMQTemplate) : ApplicationRunner {

    private val log = LoggerFactory.getLogger(javaClass)

    override fun run(args: ApplicationArguments) = runBlocking {
        for (i in 1..3) {
            // Coroutine extension from the Kotlin starter.
            template.sendEventSuspend("spring-events.basic-pubsub", "Hello KubeMQ #$i")
            log.info("Published event #{}", i)
        }
    }
}

To attach metadata, pass a tags map with the sendEvent(channel, data, tags) overload:

Publishing with tags

template.sendEvent(
        "spring-events.basic-pubsub",
        "Hello KubeMQ",
        Map.of("source", "order-service", "region", "us"));

Async publishing

For non-blocking publishing, sendEventAsync returns a CompletableFuture<Void> that completes when the broker has accepted the event. It also has a tags overload.

Async publish

CompletableFuture<Void> future =
        template.sendEventAsync("spring-events.basic-pubsub", "Hello KubeMQ");

future.thenRun(() -> log.info("Event accepted by broker"));

Subscribing to Events

Annotate a bean method with @KubeMQEventListener and accept an EventMessageReceived. The starter opens the subscription on application startup and invokes the method for each event delivered to the channel.

BasicPubSubListener.java

@Component
public class BasicPubSubListener {

    private static final Logger log = LoggerFactory.getLogger(BasicPubSubListener.class);

    @KubeMQEventListener(channels = "spring-events.basic-pubsub")
    public void onEvent(EventMessageReceived event) {
        byte[] body = event.getBody();
        String text = body != null ? new String(body, StandardCharsets.UTF_8) : "<empty>";
        log.info("Received event: channel={} body={}", event.getChannel(), text);
    }
}

The @KubeMQEventListener annotation supports these attributes. All string attributes accept SpEL expressions and ${...} property placeholders, so channels and tuning can be externalized to configuration.

Prop

Type

Consumer groups

When several listeners share the same group, the broker load-balances events across the group members instead of fanning out to all of them — exactly one member receives each event. This is how you scale out processing horizontally. The example below pairs two identical listeners in the my-group group.

ConsumerGroupListenerA.java

@Component
public class ConsumerGroupListenerA {

    private static final Logger log = LoggerFactory.getLogger(ConsumerGroupListenerA.class);

    @KubeMQEventListener(channels = "spring-events.consumer-group", group = "my-group")
    public void onEvent(EventMessageReceived event) {
        String body = event.getBody() != null
                ? new String(event.getBody(), StandardCharsets.UTF_8) : "<empty>";
        log.info("[Listener-A] Received: {}", body);
    }
}

ConsumerGroupListenerB is identical except for its log prefix and joins the same group. With both running, the events published to spring-events.consumer-group are split between the two listeners.

Multiple subscribers (fanout)

Without a group, every listener on a channel receives a copy of each event. Two beans subscribed to spring-events.multi-sub both receive every message.

SubscriberOne.java

@Component
public class SubscriberOne {

    private static final Logger log = LoggerFactory.getLogger(SubscriberOne.class);

    @KubeMQEventListener(channels = "spring-events.multi-sub")
    public void onEvent(EventMessageReceived event) {
        String body = event.getBody() != null
                ? new String(event.getBody(), StandardCharsets.UTF_8) : "<empty>";
        log.info("[Subscriber-1] Received: {}", body);
    }
}

Wildcard subscriptions

A channel pattern ending in > matches all sub-channels at and below that prefix, so one listener can consume an entire channel hierarchy. The listener below receives events published to spring-events.orders.us, spring-events.orders.eu, spring-events.orders.asia, and any other spring-events.orders.* channel.

WildcardSubscribeListener.java

@Component
public class WildcardSubscribeListener {

    private static final Logger log = LoggerFactory.getLogger(WildcardSubscribeListener.class);

    @KubeMQEventListener(channels = "spring-events.orders.>")
    public void onEvent(EventMessageReceived event) {
        byte[] body = event.getBody();
        String text = body != null ? new String(body, StandardCharsets.UTF_8) : "<empty>";
        log.info("Wildcard received: channel={} body={}", event.getChannel(), text);
    }
}

Cancelling a subscription

Annotation-driven listeners are managed by the application lifecycle, but you can also subscribe imperatively against the underlying PubSubClient and cancel the subscription when you no longer need it. Build an EventsSubscription, subscribe, and call cancel() to stop receiving events.

CancelSubscriptionRunner.java

EventsSubscription subscription = EventsSubscription.builder()
        .channel("spring-events.cancel-sub")
        .onReceiveEventCallback(event -> {
            String body = event.getBody() != null
                    ? new String(event.getBody(), StandardCharsets.UTF_8) : "<empty>";
            log.info("Received: {}", body);
        })
        .onErrorCallback(err -> log.error("Subscription error: {}", err.getMessage()))
        .build();

pubSubClient.subscribeToEvents(subscription);
log.info("Subscription started on spring-events.cancel-sub");

template.sendEvent("spring-events.cancel-sub", "Before cancel"); // received

subscription.cancel();
log.info("Subscription cancelled");

template.sendEvent("spring-events.cancel-sub", "After cancel (not received)"); // dropped

Publishing to Events Store

Publishing to Events Store is identical to publishing events, but uses sendEventStore. The broker persists each message to the channel's durable log and assigns it a sequence number, making it available for later replay.

StartNewOnlyRunner.java

@Component
public class StartNewOnlyRunner implements ApplicationRunner {

    private final KubeMQTemplate template;

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

    @Override
    public void run(ApplicationArguments args) throws InterruptedException {
        Thread.sleep(500);
        for (int i = 1; i <= 5; i++) {
            template.sendEventStore("spring-events-store.start-new-only", "New-only event #" + i);
        }
    }
}

A tags overload (sendEventStore(channel, data, tags)) and an async variant (sendEventStoreAsync, returning CompletableFuture<Void>) are available, mirroring the plain events API.

Async events-store publish

CompletableFuture<Void> future =
        template.sendEventStoreAsync("spring-events-store.start-new-only", "Hello");

Subscribing with Replay

@KubeMQEventStoreListener subscribes to a persisted channel and receives EventStoreMessageReceived instances, which expose the broker-assigned getSequence() in addition to the body and tags. The subscriptionType attribute controls where the replay begins.

Prop

Type

Start types

Each subscriptionType maps to a runnable example in the repository. The StartAtSequence, StartAtTime, and StartAtTimeDelta types additionally read subscriptionValue.

`subscriptionType`	Behavior	Example module
`StartNewOnly`	Only events published after the subscription starts.	`events-store-start-new-only`
`StartFromFirst`	Replay from the first persisted event (sequence 1).	`events-store-start-from-first`
`StartFromLast`	Deliver only the last persisted event, then continue live.	`events-store-start-from-last`
`StartAtSequence`	Replay starting at `subscriptionValue` (a sequence number).	`events-store-replay-sequence`
`StartAtTime`	Replay starting at `subscriptionValue` (an epoch timestamp).	`events-store-replay-time`
`StartAtTimeDelta`	Replay events from the last `subscriptionValue` seconds.	`events-store-time-delta`

StartNewOnlyListener.java

@Component
public class StartNewOnlyListener {

    private static final Logger log = LoggerFactory.getLogger(StartNewOnlyListener.class);

    @KubeMQEventStoreListener(
            channels = "spring-events-store.start-new-only",
            subscriptionType = "StartNewOnly")
    public void onEvent(EventStoreMessageReceived event) {
        String body = event.getBody() != null
                ? new String(event.getBody(), StandardCharsets.UTF_8) : "<empty>";
        log.info("Received (new-only): sequence={} body={}", event.getSequence(), body);
    }
}

StartFromFirstListener.java

@Component
public class StartFromFirstListener {

    private static final Logger log = LoggerFactory.getLogger(StartFromFirstListener.class);

    @KubeMQEventStoreListener(
            channels = "spring-events-store.start-from-first",
            subscriptionType = "StartFromFirst")
    public void onEvent(EventStoreMessageReceived event) {
        String body = event.getBody() != null
                ? new String(event.getBody(), StandardCharsets.UTF_8) : "<empty>";
        log.info("Received (from-first): sequence={} body={}", event.getSequence(), body);
    }
}

StartFromLastListener.java

@Component
public class StartFromLastListener {

    private static final Logger log = LoggerFactory.getLogger(StartFromLastListener.class);

    @KubeMQEventStoreListener(
            channels = "spring-events-store.start-from-last",
            subscriptionType = "StartFromLast")
    public void onEvent(EventStoreMessageReceived event) {
        String body = event.getBody() != null
                ? new String(event.getBody(), StandardCharsets.UTF_8) : "<empty>";
        log.info("Received (from-last): sequence={} body={}", event.getSequence(), body);
    }
}

ReplayFromSequenceListener.java

@Component
public class ReplayFromSequenceListener {

    private static final Logger log = LoggerFactory.getLogger(ReplayFromSequenceListener.class);

    @KubeMQEventStoreListener(
            channels = "spring-events-store.replay-sequence",
            subscriptionType = "StartAtSequence",
            subscriptionValue = "5")
    public void onEvent(EventStoreMessageReceived event) {
        String body = event.getBody() != null
                ? new String(event.getBody(), StandardCharsets.UTF_8) : "<empty>";
        log.info("Replayed from sequence: sequence={} body={}", event.getSequence(), body);
    }
}

ReplayFromTimeListener.java

@Component
public class ReplayFromTimeListener {

    private static final Logger log = LoggerFactory.getLogger(ReplayFromTimeListener.class);

    // subscriptionValue is externalized; resolves to an epoch timestamp at runtime.
    @KubeMQEventStoreListener(
            channels = "spring-events-store.replay-time",
            subscriptionType = "StartAtTime",
            subscriptionValue = "${kubemq.replay.start-time:0}")
    public void onEvent(EventStoreMessageReceived event) {
        String body = event.getBody() != null
                ? new String(event.getBody(), StandardCharsets.UTF_8) : "<empty>";
        log.info("Replayed from time: sequence={} body={}", event.getSequence(), body);
    }
}

TimeDeltaListener.java

@Component
public class TimeDeltaListener {

    private static final Logger log = LoggerFactory.getLogger(TimeDeltaListener.class);

    // Replay everything from the last 60 seconds.
    @KubeMQEventStoreListener(
            channels = "spring-events-store.time-delta",
            subscriptionType = "StartAtTimeDelta",
            subscriptionValue = "60")
    public void onEvent(EventStoreMessageReceived event) {
        String body = event.getBody() != null
                ? new String(event.getBody(), StandardCharsets.UTF_8) : "<empty>";
        log.info("Received (time-delta 60s): sequence={} body={}", event.getSequence(), body);
    }
}

Fluent Builder Alternative

For publishing, the template also exposes a fluent builder. template.newEvent(data) and template.newEventStore(data) return a KubeMQEventMessageBuilder on which you set the channel, tags, and optional metadata before calling send() (or sendAsync() for a CompletableFuture<Void>). This reads well when attaching several tags or metadata at the call site.

FluentBuildersRunner.java

// Fire-and-forget event with a tag.
template.newEvent("Tagged event")
        .toChannel("spring-fluent.events")
        .withTag("source", "fluent-builder")
        .send();

// Persisted events-store message with metadata and async send.
template.newEventStore("Audit record")
        .toChannel("spring-fluent.events-store")
        .withMetadata("v1")
        .withTag("kind", "audit")
        .sendAsync();

Kotlin Coroutines

The Kotlin starter (kubemq-spring-boot-starter-kotlin) adds suspend extension functions that bridge the template's CompletableFuture-based async methods to coroutines. sendEventSuspend and sendEventStoreSuspend suspend until the broker accepts the message and respect structured-concurrency cancellation. Each has an overload that accepts a tags Map<String, String>.

PublishService.kt

@Service
class PublishService(private val template: KubeMQTemplate) {

    suspend fun publishOrder(order: String) {
        // Fire-and-forget event.
        template.sendEventSuspend("spring-events.orders", order)

        // Persisted events-store message with tags.
        template.sendEventStoreSuspend(
            "spring-events-store.orders",
            order,
            mapOf("source" to "order-service"),
        )
    }
}

Events vs Events Store

Choose Events for real-time, ephemeral notifications where losing a message to a momentarily disconnected subscriber is acceptable — telemetry, live UI updates, presence. Choose Events Store when you need durability and replay: late-joining consumers, audit trails, or rebuilding state from history.

	Events	Events Store
Persistence	None (in-memory fanout)	Persisted to the broker
Delivery to offline subscribers	Lost	Replayable later
Replay	Not supported	Choose a start point via `subscriptionType`
Publish API	`sendEvent` / `sendEventAsync`	`sendEventStore` / `sendEventStoreAsync`
Listener	`@KubeMQEventListener`	`@KubeMQEventStoreListener`

Events & Events Store