Apache Pulsar - как ведет себя метод Consumer.seek () по метке времени?

https://pulsar.apache.org/api/client/2.4.0/org/apache/pulsar/client/api/Consumer.html#seek-long-

Должна ли временная метка при вызове метода seek (long timestamp) для потребителя равняться точному времени публикации сообщения? Например, если я отправил три сообщения в t = 1, 5, 7 и вызову consumer.seek (3), получу ли я сообщение об ошибке? или мой потребитель будет сброшен на t = 3, так что если я вызову consumer.next (), я получу второе сообщение?

Заранее спасибо,


java apache-pulsar
person pandawithcat    schedule 26.10.2020    source источник

Ответы (1)


arrow_upward
2
arrow_downward

Consumer#seek(long timestamp) позволяет сбросить подписку на заданную метку времени. После поиска потребитель начнет получать сообщения со временем публикации, равным или большим, чем временная метка, переданная методу seek.

В приведенном ниже примере показано, как сбросить потребителя на предыдущий час:

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);
    }
}

Обратите внимание, что если у вас есть несколько потребителей, принадлежащих к одному индексу (например, Key_Shared), то все потребители будут сброшены.

person fhussonnois    schedule 29.10.2020