Queues

KubeMQ queues are durable, point-to-point channels with competing consumers: every message is delivered to exactly one consumer, messages are persisted by the broker until acknowledged, and multiple instances polling the same channel share the load. Unlike Events — which the broker pushes to subscribers — the Spring adapter polls the broker for queue messages. The @KubeMQQueueListener runs one or more poll loops that pull a batch of messages each cycle, hand them to your method, and acknowledge them based on how you configure the listener.

This page covers sending, the polling listener and its tuning attributes, the acknowledgment model, advanced queue features (delay, expiration, peek, dead-letter), and the Kotlin coroutine API. For the broader queue concept, see the core Queues guide; for fire-and-forget delivery, see Events.

The template enqueues durable messages; competing @KubeMQQueueListener instances poll the queue, and each message goes to exactly one.

Sending Messages

Inject KubeMQTemplate and call sendQueueMessage(channel, data). The template serializes the payload (a byte[], a String, or any object handled by the configured message converter) and enqueues it durably. The blocking overload also accepts a Map<String, String> of tags as a third argument; sendQueueMessageAsync returns a CompletableFuture<Void> for non-blocking sends.

@Component
public class SendReceiveRunner implements ApplicationRunner {

    private final KubeMQTemplate template;

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

    @Override
    public void run(ApplicationArguments args) throws InterruptedException {
        for (int i = 1; i <= 3; i++) {
            template.sendQueueMessage("spring-queues.send-receive", "Queue message #" + i);
        }
    }
}

To send several messages to the same channel in one call, use sendQueueMessages(channel, List<?>). Each element is serialized and enqueued; messages are pre-built before sending so a serialization error fails fast before anything is published.

template.sendQueueMessages("spring-queues.batch-send",
        List.of("Batch msg #1", "Batch msg #2", "Batch msg #3"));

Consuming Messages

Annotate a bean method with @KubeMQQueueListener to start a poll loop against one or more channels. The method receives a single QueueMessageReceived; call getId() for the message id and getBody() for the raw payload bytes.

package io.kubemq.spring.boot.examples.queues;

import io.kubemq.sdk.queues.QueueMessageReceived;
import io.kubemq.spring.boot.autoconfigure.listener.KubeMQQueueListener;
import java.nio.charset.StandardCharsets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

@Component
public class SendReceiveListener {

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

    @KubeMQQueueListener(channels = "spring-queues.send-receive", autoAck = "true")
    public void onMessage(QueueMessageReceived msg) {
        byte[] body = msg.getBody();
        String text = body != null ? new String(body, StandardCharsets.UTF_8) : "<empty>";
        log.info("Received queue message: id={} body={}", msg.getId(), text);
    }
}

To process messages in batches, set batch = "true" and declare the parameter as List<QueueMessageReceived>. The listener delivers up to maxPollMessages messages per cycle in a single invocation:

@Component
public class BatchSendListener {

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

    @KubeMQQueueListener(channels = "spring-queues.batch-send", batch = "true",
            maxPollMessages = "10", autoAck = "true")
    public void onMessages(List<QueueMessageReceived> messages) {
        log.info("Received batch of {} messages", messages.size());
        for (QueueMessageReceived msg : messages) {
            String body = msg.getBody() != null
                    ? new String(msg.getBody(), StandardCharsets.UTF_8) : "<empty>";
            log.info("  Batch message: id={} body={}", msg.getId(), body);
        }
    }
}

The channels attribute supports SpEL expressions and property placeholders, so you can drive the channel name from configuration — for example channels = "${kubemq.channels.orders}".

Listener Tuning

Each @KubeMQQueueListener attribute is a String so it can carry a literal, a SpEL expression, or a ${...} placeholder. When an attribute is left blank, the listener falls back to the matching default under kubemq.listener.queues.*. The attributes and their defaults:

Set the global defaults under the kubemq.listener.queues prefix so every queue listener inherits them, then override per listener where needed:

kubemq:
  listener:
    concurrency: 1
    queues:
      poll-timeout: 5s
      max-poll-messages: 1
      visibility-timeout: 30s
      auto-ack: false

For example, a high-throughput listener that pulls ten messages per poll across four concurrent loops and acknowledges manually:

@KubeMQQueueListener(
        channels = "spring-queues.orders",
        maxPollMessages = "10",
        concurrency = "4",
        visibilityTimeout = "60",
        autoAck = "false")
public void onMessages(List<QueueMessageReceived> messages) {
    // process each message and ack/reject individually
}

Acknowledgment Model

With autoAck = "false" (the default), your method owns the outcome of every message:

• msg.ack() — the message was processed successfully; remove it from the queue.
• msg.reject() — processing failed; return the message to the queue so it can be redelivered (and eventually dead-lettered, if configured).

The ack/reject example acknowledges valid messages and rejects ones it cannot process:

@KubeMQQueueListener(channels = "spring-queues.ack-reject", autoAck = "false")
public void onMessage(QueueMessageReceived msg) {
    byte[] body = msg.getBody();
    String text = body != null ? new String(body, StandardCharsets.UTF_8) : "<empty>";
    if ("INVALID".equals(text)) {
        log.warn("Rejecting invalid message: id={}", msg.getId());
        msg.reject();
    } else {
        log.info("Processing and acknowledging: id={} body={}", msg.getId(), text);
        msg.ack();
    }
}

When you prefer to drain a queue imperatively rather than through the annotation, inject the QueuesClient directly, build a QueuesPollRequest with autoAckMessages(false), and acknowledge each returned message after processing. This "ack-all" pattern pulls a whole batch, processes it, then acks every message:

QueuesPollRequest request = QueuesPollRequest.builder()
        .channel("spring-queues.ack-all")
        .pollMaxMessages(10)
        .pollWaitTimeoutInSeconds(5)
        .autoAckMessages(false)
        .build();

QueuesPollResponse response = queuesClient.receiveQueuesMessages(request);
if (response.getMessages() != null) {
    for (QueueMessageReceived msg : response.getMessages()) {
        // process ...
        msg.ack();
    }
    log.info("Acknowledged all {} messages", response.getMessages().size());
}

Advanced Queue Features

The fluent builder (template.newQueueMessage(...), covered below) and the QueuesClient expose queue-specific behaviors beyond plain send and receive.

Error Backoff

When a poll loop throws — for example because the broker is unreachable — the listener backs off before retrying instead of spinning. The backoff starts at error-backoff-initial, multiplies by error-backoff-multiplier after each failure, and is capped at error-backoff-max:

kubemq:
  listener:
    queues:
      error-backoff-initial: 1s
      error-backoff-max: 30s
      error-backoff-multiplier: 2.0

With these defaults the retry delay grows 1s → 2s → 4s → 8s → 16s → 30s (capped) and resets once a poll succeeds.

Fluent Builder

template.newQueueMessage(data) returns a KubeMQQueueMessageBuilder for composing a message with channel, tags, metadata, and the advanced options above in one chain, then send() (blocking) or sendAsync() (returning a CompletableFuture<Void>):

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

The builder also exposes withTag(key, value), withTags(Map), withMetadata(String), and withDeadLetterQueue(channel, maxReceiveCount).

Java and Kotlin

The starter is fully usable from Java. The optional kubemq-spring-boot-starter-kotlin module adds coroutine-friendly suspend extensions on KubeMQTemplate; sendQueueMessageSuspend delegates to sendQueueMessageAsync and suspends until the send completes, honoring structured-concurrency cancellation.

import io.kubemq.spring.boot.autoconfigure.template.KubeMQTemplate;
import org.springframework.stereotype.Service;

@Service
public class OrderProducer {

    private final KubeMQTemplate template;

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

    public void enqueue(String order) {
        template.sendQueueMessage("spring-queues.orders", order);
    }
}

import io.kubemq.spring.boot.autoconfigure.template.KubeMQTemplate
import io.kubemq.spring.boot.kotlin.sendQueueMessageSuspend
import org.springframework.stereotype.Service

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

    suspend fun enqueue(order: String) {
        template.sendQueueMessageSuspend("spring-queues.orders", order)
    }
}

Running the Examples

The queue listeners and producers above need a running KubeMQ broker. The Spring Boot starter connects over gRPC on port 50000, which is also the starter's default address (localhost:50000):

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

Port 50000 is the native gRPC port the starter uses; port 9090 is the broker's shared HTTP/REST and dashboard endpoint, exposed here for convenience. The Spring Boot integration does not require the HTTP connector — it speaks gRPC directly through the KubeMQ Java SDK.

Next Steps