Introduce topic reader

[merlimat]

Goal

Introduce a new low-level API that allow applications to read through all the messages available in a topic without the need of creating a subscription.

A reader is a new entity in the Pulsar client API that will only exists when connected to a broker. Reader will be able to decide at which message id to start reading, and it will read all
the messages after that.

A reader is only useful in practice if the retention time is set to keep the data available for a given amount of time. The reader being connected will not prevent the broker to delete the data once the retention period expires.

Reader API

interface PulsarClient {
    // ......

    Reader createReader(String topic, MessageId startMessageId,
                ReaderConfiguration conf) throws PulsarClientException;

    CompletableFuture<Reader> createReaderAsync(String topic,
            MessageId startMessageId, ReaderConfiguration conf);

}

interface Reader {
    Message readNext();
    Message readNext(int timeout, TimeUnit unit);
    CompletableFuture<Message> readNextAsync();
}

Usage example:

PulsarClient client = PulsarClient.create(serviceUrl, clientConf);

ReaderConf readerConf = new ReaderConf();
Reader reader = client.createReader(topic, MessageId.earliest, readerConf);

while (true) {
    Message message = reader.readNext();

    // process message
}

API semantics

1. This API only works on "simple" topics, not on partitioned topics. One can use the Reader API to read the individual partitions.

2. Implementation of Reader will be based on the current consumer implementation, by creating a non-durable subscription on the topic and reusing the same logic on the client side.

3. In case of broker restarts, disconnections or any other transient error, the client API will resend the reposition just after the last message that was seen by the application.

4. If a reader tries to read from a message id that is before the 1st available message, it will get returned messages from the available message. Application can check the returned messages ids if it wants to double-check whether it got the data at the position it asked or some later data.

Stats

Readers will appears in stats like other subscriptions, though they will disappear after disconnection.

The subscription name will be based on a UUID and will be like reader-1231312.

Wire protocol changes

No new commands are needed, though the subscribe command needs to be extended to accept a durable boolean flag and an optional start_message_id, to manually position the subscription on a given message.

Broker changes

At the broker dispatcher level no changes will be required, except for removing non-durable subscription after consumer disconnects. In managed ledger we need to have the concept of a lightweight read cursor, that only allow for read operations (no deletes, seeks, ...). Read methods implementation should be shared with regular ManagedCursorImpl.

/cc @saandrews @rdhabalia @joefk @msb-at-yahoo

[saandrews]

Should mark delete move forward even if a reader is connected, but not reading messages?

[merlimat]

Should mark delete move forward even if a reader is connected, but not reading messages?

@saandrews What do you mean exactly? Can you give an example?

[saandrews]

Assume there is a Reader and another persistent subscription connected and both start to drain from msg L1:E1. If Reader stoped calling receive after msg L1:E10(but still connected), but the durable subscription continue to call receive to a point where we rollover the ledger and the broker can delete the ledger because it's not needed for the durable sub anymore. At this point, do we keep the ledger around since the Reader is connected and has to drain from L1:E11 or bump the reader position to the new ledger L2:E1?

[merlimat]

At this point, do we keep the ledger around since the Reader is connected and has to drain from L1:E11 or bump the reader position to the new ledger L2:E1?

In the PR I've submitted the data get deleted as if the TopicReader was not there. Internally, the topic reader has an in-memory only mark-delete position, that is used to expose in the stats.

My argument for not making the topic reader retain messages is that, if there is a disconnection, the topic reader is gone and the data will get deleted anyway. The mark-delete position will not immediately advance, but the next message will be the next currently available message.

If you want to use the topic reader (which is kind of a lower level API), you need to ensure the data is retained for a given amount of time, by configuring a non-zero retention time.

[merlimat]

.. or bump the reader position to the new ledger L2:E1

so, yes, the reader will be bumped on L2:E1 the next time it tries to read.

[saandrews]

OK, Thanks. 👍

[merlimat]

Merged in #371