Requeue All
Return all received KubeMQ Queue messages to the queue with the Java SDK so they can be redelivered to consumers.
Overview
Requeue all moves an entire batch of polled messages to a different channel in one server-side operation, without republishing them from the client. Reach for it when you need to make a routing decision after looking at a batch — shovel a stuck batch into a review queue, redirect it to a priority pipeline, or migrate messages off a channel that's being retired, all while the source queue is cleared atomically.
It works against the response returned by a manual poll: after receiving messages, call response.reQueueAll(destinationChannel) to move every message in that response to the destination channel in one operation, removing them from the source at the same instant. The messages keep their original body, tags, and policies — the broker relocates them, it doesn't recreate them.
Gotchas: requeuing is all-or-nothing for the batch — there's no per-message filter, so split the batch yourself first if only some messages should move. The destination channel is an ordinary queue with no special semantics; nothing consumes it automatically. And the operation only affects messages still held from that poll — anything already acked or expired beforehand is gone before reQueueAll runs.
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;
/**
* RequeueAllExample for Queues Stream
*/
public class RequeueAllExample {
private static final String ADDRESS = "localhost:50000";
private static final String CLIENT_ID = "java-queues-requeue-all-client";
private static final String CHANNEL = "java-queues.requeue-all";
private static final String REQUEUE_CHANNEL = "java-queues.requeue-all-dest";
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 source and destination queue channels
client.createQueuesChannel(CHANNEL);
client.createQueuesChannel(REQUEUE_CHANNEL);
// Send messages to the source queue
for (int i = 1; i <= 3; i++) {
client.sendQueueMessage(QueueMessage.builder()
.channel(CHANNEL).body(("Requeue msg " + i).getBytes()).build());
}
// Poll for messages from the source queue
QueuesPollResponse response = client.receiveQueueMessages(QueuesPollRequest.builder()
.channel(CHANNEL).pollMaxMessages(10).pollWaitTimeoutInSeconds(5).build());
System.out.println("Received " + response.getMessages().size() + " messages.");
// Requeue all messages to a different channel
response.reQueueAll(REQUEUE_CHANNEL);
System.out.println("All messages requeued to: " + REQUEUE_CHANNEL);
// Verify messages arrived in the destination queue
QueuesPollResponse dest = client.receiveQueueMessages(QueuesPollRequest.builder()
.channel(REQUEUE_CHANNEL).pollMaxMessages(10).pollWaitTimeoutInSeconds(2).autoAckMessages(true).build());
System.out.println("Destination queue received: " + dest.getMessages().size() + " messages.");
// Clean up resources
client.deleteQueuesChannel(CHANNEL);
client.deleteQueuesChannel(REQUEUE_CHANNEL);
} catch (Exception e) {
System.err.println("Error: " + e.getMessage());
}
}
}
How It Works
response.reQueueAll(REQUEUE_CHANNEL)atomically moves all polled messages to the destination channel in one operation; the source messages are removed from the original queue.- The destination channel
java-queues.requeue-all-destis a separate queue that can be consumed independently, enabling routing-on-failure patterns without manual re-publish loops. - A final poll on
REQUEUE_CHANNELwithautoAckMessages(true)confirms all three messages arrived, verifying the requeue succeeded.
Related
Was this page helpful?