Nack All
Reject all received KubeMQ queue messages for redelivery with nackAllQueuesMessages in Kotlin.
Overview
Bulk nack rejects an entire polled batch of queue messages in a single call instead of settling each one individually. It's the operation you reach for when a failure affects the whole batch at once — a downstream dependency is down, a shared resource lock couldn't be acquired, or a transient error means none of the messages can be processed right now — and retrying them one-by-one would just be extra round-trips for the same outcome.
It works with manual-ack polling: client.receiveQueuesMessages with autoAck = false returns the batch without settling it, and client.nackAllQueuesMessages(response) sends one bulk-reject that settles every message in that response, returning them all to the queue for redelivery.
Gotchas: the receive count increments on every message in the batch, so an unbounded retry loop is one bad nackAllQueuesMessages call away — combine it with a DLQ policy to cap retries. It's all-or-nothing: you can't use it to keep a few messages and reject the rest — that needs per-message settlement. Calling it on an empty response is a wasted round-trip.
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.queueMessage
import kotlinx.coroutines.runBlocking
private const val ADDRESS = "localhost:50000"
private const val CLIENT_ID = "kotlin-queues-nack-all"
private const val CHANNEL = "kotlin-queues.nack-all"
fun main() = runBlocking {
val client = KubeMQClient.queues {
address = ADDRESS
clientId = CLIENT_ID
}
client.use {
try {
client.createQueuesChannel(CHANNEL)
// Send messages
repeat(3) { i ->
client.sendQueuesMessage(queueMessage {
channel = CHANNEL
body = "Nack msg ${i + 1}".toByteArray()
})
}
// Poll for messages
val response = client.receiveQueuesMessages {
channel = CHANNEL
maxItems = 10
waitTimeoutMs = 5000
autoAck = false
}
println("Received ${response.messages.size} messages.")
// Reject all messages (return them to queue for redelivery)
client.nackAllQueuesMessages(response)
println("All messages rejected via nackAll().")
println("Messages returned to queue for redelivery.")
// Cleanup: consume rejected messages
val cleanup = client.receiveQueuesMessages {
channel = CHANNEL
maxItems = 10
waitTimeoutMs = 1000
autoAck = true
}
println("Cleanup: ${cleanup.messages.size} messages consumed.")
} finally {
try { client.deleteQueuesChannel(CHANNEL) } catch (_: Exception) {}
}
println("Done.")
}
}How It Works
nackAllQueuesMessages(response)rejects all messages in the batch, returning them to the queue.- Rejected messages become available for redelivery to other consumers.
- Useful when a batch processing operation fails and all messages need to be retried.
- Combine with DLQ policy to limit retry attempts.
Related
Was this page helpful?