Ack & Reject
Selectively acknowledge or reject individual KubeMQ Queue messages with the Java SDK to control redelivery.
Overview
Ack and reject give you per-message control over queue delivery instead of an all-or-nothing batch outcome. When receiveQueueMessages fetches a batch, 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 QueueMessageReceived: msg.ack(), which permanently removes it from the queue, and msg.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 — an uncleared rejected message will keep reappearing on subsequent polls.
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;
/**
* Ack/Reject Example
*
* Demonstrates acknowledging and rejecting individual messages based on processing outcome.
*/
public class AckRejectExample {
private static final String ADDRESS = "localhost:50000";
private static final String CLIENT_ID = "java-queues-ack-reject-client";
private static final String CHANNEL = "java-queues.ack-reject";
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 <= 3; i++) {
client.sendQueueMessage(QueueMessage.builder()
.id(UUID.randomUUID().toString()).channel(CHANNEL)
.body(("Message " + i).getBytes()).build());
}
System.out.println("Sent 3 messages.\n");
// Receive messages from the queue
QueuesPollResponse response = client.receiveQueueMessages(QueuesPollRequest.builder()
.channel(CHANNEL).pollMaxMessages(10).pollWaitTimeoutInSeconds(5).build());
// Handle each message: ack on success, reject on failure
for (QueueMessageReceived msg : response.getMessages()) {
String body = new String(msg.getBody());
if (body.contains("2")) {
msg.reject();
System.out.println(" REJECTED: " + body);
} else {
msg.ack();
System.out.println(" ACKNOWLEDGED: " + body);
}
}
// Clean up rejected message (receive and auto-ack)
QueuesPollResponse cleanup = client.receiveQueueMessages(QueuesPollRequest.builder()
.channel(CHANNEL).pollMaxMessages(10).pollWaitTimeoutInSeconds(1).autoAckMessages(true).build());
System.out.println("\nRemaining messages cleaned up: " + cleanup.getMessages().size());
// Clean up resources
client.deleteQueuesChannel(CHANNEL);
} catch (Exception e) {
System.err.println("Error: " + e.getMessage());
}
}
}
How It Works
msg.ack()removes the message from the queue permanently;msg.reject()returns it to the queue so another consumer can try again.- The example rejects message #2 (body contains "2") and acknowledges messages #1 and #3 to demonstrate per-message decision logic.
- A cleanup poll with
autoAckMessages(true)drains the rejected message so the channel can be deleted cleanly; in production you would route rejected messages to a dead-letter queue instead.
Related
Was this page helpful?