Ack Range

Overview

A single poll response often bundles several messages into one batch, but "successfully processed" rarely applies to all of them uniformly — one handler might fail while its siblings succeed. Settling the whole batch together forces an all-or-nothing outcome: either you redeliver work you already finished, or you silently drop work you didn't. Per-message settlement lets each message's outcome reflect what actually happened to it, instead of the worst result in the batch.

Each QueueReceivedMessage returned by receiveQueuesMessages carries its own broker-assigned attributes.sequence. Calling msg.ack() on one message settles only that message; calling msg.reject() on another explicitly returns it to the queue for redelivery. Messages you don't touch at all are left pending — still redeliverable — until their own ack()/reject() is called or the visibility timeout expires.

Gotchas: messages you never touch aren't automatically fine — once the visibility timeout elapses, anything left unsettled goes back to the queue, so a handler that forgets to settle a message isn't "done," it's "will retry." Every reject() increments that message's attributes.receiveCount, which can trigger dead-letter routing if maxReceiveCount is configured — so a bug that rejects indiscriminately can drain a message's retry budget fast. And selective settlement only works with messages fetched with autoAck = false — with auto-ack on, the broker settles the entire batch the moment it's delivered, before your code runs.

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-ack-range"
private const val CHANNEL = "kotlin-queues.ack-range"

fun main() = runBlocking {
    val client = KubeMQClient.queues {
        address = ADDRESS
        clientId = CLIENT_ID
    }

    client.use {
        try {
            client.createQueuesChannel(CHANNEL)

            // Send messages
            repeat(5) { i ->
                client.sendQueuesMessage(queueMessage {
                    channel = CHANNEL
                    body = "Range msg ${i + 1}".toByteArray()
                })
            }

            // Poll for a batch of messages with manual ack
            val response = client.receiveQueuesMessages {
                channel = CHANNEL
                maxItems = 5
                waitTimeoutMs = 5000
                autoAck = false
            }

            println("Received ${response.messages.size} messages:")

            // Selectively settle by sequence: ack even sequences, reject odd ones
            for (msg in response.messages) {
                val sequence = msg.attributes.sequence
                if (sequence % 2 == 0L) {
                    msg.ack()
                    println("  Acked: seq=$sequence")
                } else {
                    msg.reject()
                    println("  Rejected: seq=$sequence")
                }
            }

            // Cleanup rejected messages
            val cleanup = client.receiveQueuesMessages {
                channel = CHANNEL
                maxItems = 10
                waitTimeoutMs = 1000
                autoAck = true
            }
            println("\nCleanup: consumed ${cleanup.messages.size} rejected messages.")
        } finally {
            try { client.deleteQueuesChannel(CHANNEL) } catch (_: Exception) {}
        }
        println("Done.")
    }
}

How It Works

• receiveQueuesMessages polls with autoAck = false, so each QueueReceivedMessage must be settled explicitly via msg.ack() or msg.reject().
• Settlement decisions are made per-message using the broker-assigned msg.attributes.sequence (even = ack, odd = reject in this example).
• Rejected messages are returned to the queue for redelivery; their attributes.receiveCount increments, which can trigger dead-letter routing if maxReceiveCount is set.
• This per-message settlement pattern is the foundation for selective processing: only confirmed-successful messages are removed from the queue, while the rest remain redeliverable.

Related

• Ack All
• Nack All
• Queues Reference