Apache Pulsar - What is the behaviour of the Consumer.seek() method by timestamp?

Viewed 600
1 Answers

The Consumer#seek(long timestamp) allows you to reset your subscription to a given timestamp. After seeking the consumer will start receiving messages with a publish time equal to or greater than the timestamp passed to the seek method.

The below example show how to reset a consumer to the previous hour:

try (
    // Create PulsarClient
    PulsarClient client = PulsarClient
        .builder()
        .serviceUrl("pulsar://localhost:6650")
        .build();
    // Create Consumer subscription
    Consumer<String> consumer = client.newConsumer(Schema.STRING)
        .topic("my-topic")
        .subscriptionName("my-subscription")
        .subscriptionMode(SubscriptionMode.Durable)
        .subscriptionType(SubscriptionType.Key_Shared)
        .subscriptionInitialPosition(SubscriptionInitialPosition.Latest)
        .subscribe()
) {
    // Seek consumer to previous hour
    consumer.seek(Instant.now().minus( Duration.ofHours(1)).toEpochMilli());
    while (true) {
        final Message<String> msg = consumer.receive();
        System.out.printf(
            "Message received: key=%s, value=%s, topic=%s, id=%s%n",
            msg.getKey(),
            msg.getValue(),
            msg.getTopicName(),
            msg.getMessageId().toString());
        consumer.acknowledge(msg);
    }
}

Note that if you have multiple consumers that belong to the same subscriptio ( e.g., Key_Shared) then all consumers will be reset.

Related