Ack All
Acknowledge all received KubeMQ Queue messages at once with the Java SDK to confirm a full batch in one call.
Which to use
This page is about settling an entire polled batch in one call when every message in it was fully handled. For selective/per-message settlement of a batch — acking or rejecting individual messages by sequence while leaving others pending — see Ack Range.
Overview
response.ackAll() acknowledges every message in a poll response with a single call, instead of walking the batch and calling msg.ack() on each one. Reach for it whenever a poll returns a batch you're confident you've fully handled — after processing all the messages in a pollMaxMessages 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 receiveQueueMessages — 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, use per-message ack()/reject() instead. Messages must have been received with manual acknowledgement (not auto-ack) for ackAll() to have anything to confirm. For unconditional channel-wide draining regardless of what's been polled, see a dead-letter policy or purge instead — ackAll() only ever operates on messages you've already received.
Prerequisites
- KubeMQ server running on
localhost:50000 - Java SDK installed (
implementation 'io.kubemq.sdk:kubemq-sdk-Java:3.1.1'(Gradle) or Maven dependency from Getting Started)
Code
package io.kubemq.example.queues;
import io.kubemq.sdk.queues.*;
import java.util.UUID;
/**
* AckAll Example
*
* Demonstrates acknowledging all messages in a single poll response using ackAll().
*/
public class AckAllExample {
private static final String ADDRESS = "localhost:50000";
private static final String CLIENT_ID = "java-queues-ack-all-client";
private static final String CHANNEL = "java-queues.ack-all";
public static void main(String[] args) {
// Create a queues client connected to the KubeMQ server
try (QueuesClient client = QueuesClient.builder().address(ADDRESS).clientId(CLIENT_ID).build()) {
// Create the queue channel
client.createQueuesChannel(CHANNEL);
// Send messages to the queue
for (int i = 1; i <= 5; i++) {
client.sendQueueMessage(QueueMessage.builder()
.id(UUID.randomUUID().toString()).channel(CHANNEL)
.body(("Message " + i).getBytes()).build());
}
System.out.println("Sent 5 messages.\n");
// Receive messages from the queue
QueuesPollResponse response = client.receiveQueueMessages(QueuesPollRequest.builder()
.channel(CHANNEL).pollMaxMessages(10).pollWaitTimeoutInSeconds(5).build());
if (!response.isError()) {
System.out.println("Received " + response.getMessages().size() + " messages.");
response.getMessages().forEach(msg ->
System.out.println(" " + new String(msg.getBody())));
// Acknowledge all messages in the response at once
response.ackAll();
System.out.println("\nAll messages acknowledged with ackAll().");
}
// Clean up resources
client.deleteQueuesChannel(CHANNEL);
} catch (Exception e) {
System.err.println("Error: " + e.getMessage());
}
}
}
How It Works
response.ackAll()sends a single bulk-acknowledge for all messages in theQueuesPollResponse, which is more efficient than callingmsg.ack()per message.pollMaxMessages(10)requests up to 10 messages in one call; all five arrive in a single response, soackAll()commits them all at once.- Messages not acknowledged within the broker's visibility timeout are returned to the queue for redelivery —
ackAll()prevents that for the entire batch.
Related
Was this page helpful?