Overview

Ack and reject give you per-message control over queue delivery instead of an all-or-nothing batch outcome. When receiveQueuesMessages fetches a batch with autoAck = false, each message stays locked on the broker — invisible to other consumers — until the consumer explicitly settles it. That's what you need when one bad record in a batch shouldn't take the rest down with it.

Settlement happens through two calls on the QueueReceivedMessage: ack(), which permanently removes it from the queue, and reject(), which returns it to the queue for redelivery. Internally the broker tracks this against a receive count, which a dead-letter policy can use to stop retrying a poison message forever.

Gotchas: an unsettled message isn't gone — it snaps back to the queue once the visibility timeout expires, so a slow consumer looks identical to a rejecting one; settle every message before that deadline, and never assume a batch is fully processed until you've called ack() or reject() on each one individually.

Prerequisites

KubeMQ server running on localhost:50000
Kotlin SDK installed (implementation("io.kubemq.sdk:kubemq-sdk-kotlin:1.0.1"))

Code

AckRejectExample.kt

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

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

    client.use {
        // Send multiple messages
        val bodies = listOf("important-task", "bad-task", "another-good-task")
        bodies.forEach { body ->
            client.sendQueuesMessage(queueMessage {
                channel = CHANNEL
                this.body = body.toByteArray()
            })
            println("Sent: $body")
        }

        // Receive and selectively ack/reject
        val response = client.receiveQueuesMessages {
            channel = CHANNEL
            maxItems = 10
            waitTimeoutMs = 5000
            autoAck = false
        }

        for (msg in response.messages) {
            val body = String(msg.body)
            if (body.contains("bad")) {
                msg.reject()
                println("Rejected: $body")
            } else {
                msg.ack()
                println("Acked: $body")
            }
        }

        // Cleanup rejected messages
        val cleanup = client.receiveQueuesMessages {
            channel = CHANNEL
            maxItems = 10
            waitTimeoutMs = 1000
            autoAck = true
        }
        println("Cleanup: consumed ${cleanup.messages.size} rejected messages.")

        println("Done.")
    }
}

How It Works

Each QueueReceivedMessage provides ack() and reject() methods for individual message handling.
Rejected messages are returned to the queue for redelivery to another consumer.
This pattern is useful for content-based filtering or validation before processing.

Ack & Reject

Overview

Prerequisites

Code

How It Works

On this page

Ack & Reject

Overview

Prerequisites

Code

How It Works

Related

On this page