Expiration Policy
Send KubeMQ queue messages with a TTL expiration policy that auto-removes expired messages, in Kotlin.
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, QueueMessagePolicy(expirationSeconds = 3) attaches a per-message TTL when you build the message, and the clock starts the moment the broker accepts it via sendQueuesMessage, 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 - Kotlin SDK installed (
implementation("io.kubemq.sdk:kubemq-sdk-kotlin:1.0.1"))
Code
package io.kubemq.sdk.examples.queuesstream
import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.queues.QueueMessagePolicy
import io.kubemq.sdk.queues.queueMessage
import kotlinx.coroutines.delay
import kotlinx.coroutines.runBlocking
private const val ADDRESS = "localhost:50000"
private const val CLIENT_ID = "kotlin-queues-expiration-policy"
private const val CHANNEL = "kotlin-queues.expiration-policy"
fun main() = runBlocking {
val client = KubeMQClient.queues {
address = ADDRESS
clientId = CLIENT_ID
}
client.use {
try {
client.createQueuesChannel(CHANNEL)
// Send message with 3s expiration
val expirationSeconds = 3
client.sendQueuesMessage(queueMessage {
channel = CHANNEL
body = "Expiring message".toByteArray()
policy = QueueMessagePolicy(expirationSeconds = expirationSeconds)
})
println("Sent message with ${expirationSeconds}s expiration.")
// Wait for message to expire
println("Waiting ${expirationSeconds + 2}s for expiration...")
delay(((expirationSeconds + 2) * 1000).toLong())
// Poll after expiration (expect no messages)
val response = client.receiveQueuesMessages {
channel = CHANNEL
maxItems = 1
waitTimeoutMs = 1000
autoAck = true
}
println("Messages after expiration: ${response.messages.size} (expected 0)")
} finally {
try { client.deleteQueuesChannel(CHANNEL) } catch (_: Exception) {}
}
println("Done.")
}
}How It Works
QueueMessagePolicy(expirationSeconds = 3)sets a 3-second TTL on the message.- After the TTL expires, the message is automatically removed from the queue.
- Polling after expiration returns no messages.
- Useful for time-sensitive data, session tokens, or temporary notifications.
Related
Was this page helpful?