Expiration Policy
Set a message expiration TTL on a KubeMQ Queue stream with the Java SDK so stale messages are dropped automatically.
Overview
An expiration policy puts a hard time limit on how long a queue message may sit unconsumed. It solves a different problem than a dead-letter policy — this isn't about messages that fail processing, it's about messages that go stale: a price quote, a one-time code, a cache-invalidation signal, where late delivery is actively wrong, not just delayed. Instead of every consumer re-checking timestamps itself, the deadline lives on the message and the broker enforces it.
At the API level, .expirationInSeconds(3) attaches a per-message TTL when you build the QueueMessage, and the clock starts the moment the broker accepts it via sendQueueMessage, not when a consumer picks it up. Let the TTL elapse unconsumed and the broker silently removes it — a later poll just comes back empty, no error, no trace.
Gotchas: expiration is silent — no DLQ routing, no event, just a message that vanishes — so pair it with monitoring if you need visibility into how much work is being dropped. The timer starts at send time, not when a consumer picks up the work, so a message can expire mid-backlog even while a consumer is actively polling. And setting the TTL too short for your real consumer lag just turns ordinary slowness into silent data loss.
Prerequisites
- KubeMQ server running on
localhost:50000 - Java SDK installed (
implementation 'io.kubemq.sdk:kubemq-sdk-Java:3.1.1'(Gradle) or Maven dependency from Getting Started)
Code
package io.kubemq.example.queuesstream;
import io.kubemq.sdk.queues.*;
import java.util.UUID;
/**
* ExpirationPolicyExample for Queues Stream
*/
public class ExpirationPolicyExample {
private static final String ADDRESS = "localhost:50000";
private static final String CLIENT_ID = "java-queues-expiration-policy-client";
private static final String CHANNEL = "java-queues.expiration-policy";
public static void main(String[] args) throws InterruptedException {
// Create a client connected to the KubeMQ server
try (QueuesClient client = QueuesClient.builder().address(ADDRESS).clientId(CLIENT_ID).build()) {
// Create the queue channel
client.createQueuesChannel(CHANNEL);
// Send a message with expiration policy (message expires after 3 seconds)
int expirationSeconds = 3;
client.sendQueueMessage(QueueMessage.builder()
.id(UUID.randomUUID().toString()).channel(CHANNEL)
.body("Expiring message".getBytes())
.expirationInSeconds(expirationSeconds).build());
System.out.println("Sent message with " + expirationSeconds + "s expiration.");
// Wait for message to expire
Thread.sleep((expirationSeconds + 2) * 1000);
// Poll after expiration (expect no messages)
QueuesPollResponse response = client.receiveQueueMessages(QueuesPollRequest.builder()
.channel(CHANNEL).pollMaxMessages(1).pollWaitTimeoutInSeconds(1).autoAckMessages(true).build());
System.out.println("Messages after expiration: " + response.getMessages().size() + " (expected 0)");
// Clean up resources
client.deleteQueuesChannel(CHANNEL);
}
}
}
How It Works
expirationInSeconds(3)is a per-message TTL; the broker discards the message if no consumer polls within 3 seconds.Thread.sleep((expirationSeconds + 2) * 1000)waits 5 seconds (3s TTL + 2s margin) before polling; the queue is empty, confirming the message expired and was not delivered.- Expiration is set per-message at publish time, independent of channel-level retention settings.
Related
Was this page helpful?