Ack All

Overview

ackAllQueuesMessages acknowledges every message in a poll response with a single call, instead of walking the batch and acking each message individually. Reach for it whenever a poll returns a batch you're confident you've fully handled — after processing all the messages in a maxItems batch, or clearing a backlog of stale work after a bad deploy — where per-message acks would just be repetitive round-trips against the same response object.

Because it settles the whole batch in one shot, it's cheaper than looping over messages, and it prevents any message in that batch from being redelivered once the broker's visibility timeout elapses. It only affects the messages already in hand from receiveQueuesMessages — it doesn't reach into the channel for anything not yet polled.

Gotchas: this is all-or-nothing for the batch — you can't ack most of a response and leave a few pending; if even one message failed processing, ack or nack messages individually instead. Messages must have been received with autoAck = false for ackAllQueuesMessages to have anything left to confirm. For unconditional channel-wide draining regardless of what's been polled, see a dead-letter policy or Requeue All instead — ackAllQueuesMessages only ever operates on messages you've already received.

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.queues

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

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

    client.use {
        // Send batch of messages
        val messages = (1..5).map { i ->
            queueMessage {
                channel = CHANNEL
                body = "Batch msg #$i".toByteArray()
            }
        }
        client.sendQueueMessagesBatch(messages)
        println("Sent ${messages.size} messages.")

        // Receive with manual ack
        val response = client.receiveQueuesMessages {
            channel = CHANNEL
            maxItems = 10
            waitTimeoutMs = 5000
            autoAck = false
        }

        println("Received ${response.messages.size} messages:")
        response.messages.forEach { println("  ${String(it.body)}") }

        // Ack all in one call
        if (response.messages.isNotEmpty()) {
            client.ackAllQueuesMessages(response)
            println("\nAcked all ${response.messages.size} messages at once.")
        }

        println("Done.")
    }
}

How It Works

• ackAllQueuesMessages(response) acknowledges all messages in the poll response in a single operation.
• This is more efficient than acknowledging each message individually when batch processing.
• Messages must be received with autoAck = false for manual acknowledgement.

Related

• Send & Receive
• Nack All
• Ack Range