Replay from Time
Replay persisted KubeMQ Events Store messages starting from a specific timestamp using the Java SDK.
Overview
Replaying from a timestamp lets a consumer recover a window of history without knowing exact sequence numbers — you reach for it after a deploy, an outage, or any gap where you know roughly when you went dark but not where you left off in the stream. It turns an Events Store channel into a rewindable log: resubscribe with a point in time and the broker replays every event stored at or after it, then hands off to live delivery.
The subscription's eventsStoreType is set to EventsStoreType.StartAtTime with .eventsStoreStartTime(startTime) given an Instant value — the broker compares this against the storage timestamp it assigned to each event, not any timestamp embedded in the payload. Because it's wall-clock based, the window is approximate rather than exact: pass a time far enough back to be safe.
Gotchas: clock skew between your subscriber's clock and the server's matters — favor a generous buffer over a precise cutoff. Storage timestamps reflect when the broker persisted the event, not when the producer created it, so under load the two can drift. And unlike sequence-based replay, a time-based start position has no way to guarantee "no gaps, no duplicates" across a network hiccup — use EventsStoreType.StartAtSequence instead if you need exact resumption.
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.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Replay From Time Example
*
* Demonstrates StartAtTime subscription that replays from a specific timestamp.
*/
public class ReplayFromTimeExample {
private static final String ADDRESS = "localhost:50000";
private static final String CLIENT_ID = "java-eventsstore-replay-from-time-client";
private static final String CHANNEL = "java-eventsstore.replay-from-time";
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 <= 5; i++) {
client.publishEventStore(EventStoreMessage.builder()
.id("msg-" + i).channel(CHANNEL)
.body(("Event #" + i).getBytes()).build());
Thread.sleep(100);
}
System.out.println("Sent 5 messages.\n");
Instant startTime = Instant.now().minus(1, ChronoUnit.HOURS);
AtomicInteger received = new AtomicInteger(0);
CountDownLatch latch = new CountDownLatch(5);
EventsStoreSubscription sub = EventsStoreSubscription.builder()
.channel(CHANNEL)
.eventsStoreType(EventsStoreType.StartAtTime)
.eventsStoreStartTime(startTime)
.onReceiveEventCallback(e -> {
received.incrementAndGet();
System.out.println(" Received: " + new String(e.getBody()));
latch.countDown();
})
.onErrorCallback(err -> System.err.println("Error: " + err))
.build();
// Subscribe with StartAtTime (replay from specific timestamp)
client.subscribeToEventsStore(sub);
System.out.println("Subscribed with StartAtTime (1 hour ago).");
latch.await(10, TimeUnit.SECONDS);
System.out.println("\nReceived " + received.get() + " messages.");
// Clean up resources
sub.cancel();
client.deleteEventsStoreChannel(CHANNEL);
client.close();
}
}
How It Works
EventsStoreType.StartAtTimewith.eventsStoreStartTime(startTime)tells the broker to replay all events stored at or after the givenInstant.startTimeis set to one hour ago, which is before any of the five messages were published, so the subscriber replays the full history.- The
Instantis passed directly; the Java SDK converts it to the gRPC timestamp type internally.
Related
Was this page helpful?