Reject All
Negative-acknowledge (reject) all received KubeMQ Queue messages with the Java SDK to trigger redelivery of the full batch.
Overview
Bulk reject (this SDK's nack) rejects an entire polled batch of queue messages in a single call instead of settling each one individually. It's the operation you reach for when a failure affects the whole batch at once — a downstream dependency is down, a shared resource lock couldn't be acquired, or a transient error means none of the messages can be processed right now — and retrying them one-by-one would just be extra round-trips for the same outcome.
Naming note: the Java SDK's negative-acknowledgment method is rejectAll() — there is no nackAll() alias. It returns every message in the batch for redelivery, the same outcome other SDKs call "nack."
It works with manual-ack polling: client.receiveQueueMessages returns a QueuesPollResponse holding the messages without settling them, and response.rejectAll() sends one bulk-reject that settles every message in that response, returning them all to the queue for redelivery.
Gotchas: the receive count increments on every message in the batch, so an unbounded retry loop is one bad rejectAll() away — pair it with a max-receive-count and a dead-letter policy. rejectAll() is all-or-nothing and is the complement to ackAll(): use ackAll() when the whole batch succeeded, rejectAll() when none of it can be processed — you can't use it to keep a few messages and reject the rest, that needs per-message settlement. Calling it on an empty poll result is a wasted round-trip.
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.queuesstream;
import io.kubemq.sdk.queues.*;
import java.util.UUID;
/**
* NackAllExample for Queues Stream
*/
public class NackAllExample {
private static final String ADDRESS = "localhost:50000";
private static final String CLIENT_ID = "java-queues-nack-all-client";
private static final String CHANNEL = "java-queues.nack-all";
/**
* Demonstrates rejecting all messages in a poll response using rejectAll().
*/
public static void main(String[] args) {
// Create a 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()
.channel(CHANNEL).body(("Nack msg " + i).getBytes()).build());
}
// Poll for messages
QueuesPollResponse response = client.receiveQueueMessages(QueuesPollRequest.builder()
.channel(CHANNEL).pollMaxMessages(10).pollWaitTimeoutInSeconds(5).build());
System.out.println("Received " + response.getMessages().size() + " messages.");
// Reject all messages (return them to queue for redelivery)
response.rejectAll();
System.out.println("All messages rejected via rejectAll().");
System.out.println("Messages returned to queue for redelivery.");
// Clean up: consume rejected messages so we can delete the channel
QueuesPollResponse cleanup = client.receiveQueueMessages(QueuesPollRequest.builder()
.channel(CHANNEL).pollMaxMessages(10).pollWaitTimeoutInSeconds(1).autoAckMessages(true).build());
System.out.println("Cleanup: " + cleanup.getMessages().size() + " messages consumed.");
// Clean up resources
client.deleteQueuesChannel(CHANNEL);
} catch (Exception e) {
System.err.println("Error: " + e.getMessage());
}
}
}
How It Works
response.rejectAll()sends a single bulk-reject for all messages in the poll response; each message is returned to the queue for redelivery.- A second poll with
autoAckMessages(true)drains the redelivered messages so the channel can be deleted cleanly; in production a redelivery count or DLQ would handle repeated failures. rejectAll()is the complement toackAll(): useackAll()when all messages in a batch succeeded,rejectAll()when a transient error means you cannot process any of them.
Related
Was this page helpful?