Replay from Sequence
Replay events starting from a specific sequence number
Overview
Replaying from a sequence number lets a consumer resume an events-store subscription from an exact point in a channel's history, instead of re-reading everything or only catching new traffic. It's the checkpoint-recovery pattern: a worker persists the last sequence it processed, and after a crash or redeploy it reopens the subscription right there — no gap, no reprocessing everything that came before.
Sequence numbers are broker-assigned per channel, starting at 1 and increasing monotonically with every stored event; they never reset unless the channel is purged. Setting .eventsStoreType(EventsStoreType.StartAtSequence) with .eventsStoreSequenceValue(5) tells the broker to begin delivery at that sequence inclusive, replaying stored events from that point, then transitioning the subscription to live delivery for anything published afterward.
Gotchas: the sequence value is inclusive, so .eventsStoreSequenceValue(5) still delivers event 5 — off by one and you'll reprocess or silently drop a message; you must track and persist the "last processed" sequence yourself, KubeMQ doesn't checkpoint it for you; and requesting a sequence past the current head isn't an error — you'll just get nothing until new events catch up to it.
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.eventsstore;
import io.kubemq.sdk.pubsub.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Replay From Sequence Example
*
* Demonstrates StartAtSequence subscription that replays from a specific sequence number.
*/
public class ReplayFromSequenceExample {
private static final String ADDRESS = "localhost:50000";
private static final String CLIENT_ID = "java-eventsstore-replay-from-sequence-client";
private static final String CHANNEL = "java-eventsstore.replay-from-sequence";
public static void main(String[] args) throws InterruptedException {
// Create a client connected to the KubeMQ server
PubSubClient client = PubSubClient.builder().address(ADDRESS).clientId(CLIENT_ID).build();
client.ping();
// Create the events store channel
client.createEventsStoreChannel(CHANNEL);
// Send messages to create history
for (int i = 1; i <= 10; i++) {
client.publishEventStore(EventStoreMessage.builder()
.id("msg-" + i).channel(CHANNEL)
.body(("Message #" + i).getBytes()).build());
}
System.out.println("Sent 10 messages.\n");
int startSequence = 5;
AtomicInteger received = new AtomicInteger(0);
CountDownLatch latch = new CountDownLatch(6);
EventsStoreSubscription sub = EventsStoreSubscription.builder()
.channel(CHANNEL)
.eventsStoreType(EventsStoreType.StartAtSequence)
.eventsStoreSequenceValue(startSequence)
.onReceiveEventCallback(e -> {
int n = received.incrementAndGet();
System.out.println(" [" + n + "] Seq " + e.getSequence() + ": " + new String(e.getBody()));
latch.countDown();
})
.onErrorCallback(err -> System.err.println("Error: " + err))
.build();
// Subscribe with StartAtSequence (replay from specific sequence number)
client.subscribeToEventsStore(sub);
System.out.println("Subscribed at sequence " + startSequence + ".");
latch.await(10, TimeUnit.SECONDS);
System.out.println("\nReceived " + received.get() + " messages (starting from seq " + startSequence + ").");
// Clean up resources
sub.cancel();
client.deleteEventsStoreChannel(CHANNEL);
client.close();
}
}
How It Works
- Ten messages are published first so the store has a known history.
EventsStoreType.StartAtSequencecombined with.eventsStoreSequenceValue(5)tells the broker to replay from sequence 5 onwards; messages 1–4 are skipped.- The latch is initialised to 6 (sequences 5–10 inclusive); the callback prints the broker-assigned sequence number from
e.getSequence()so the replay range can be verified.
Related
Was this page helpful?