Allow reader to detect when it's caught up with writers

[merlimat]

Context

We have recently introduced the concept of topic "Reader" as an alternative to the traditional pub-sub consumer abstraction.

A common followup request has been to have a way to identify when the reader has reached the last published entry on the topic.

There is no currently direct way to achieve that and using readNext(timeout) doesn't help because in case the client is not connected to broker, it doesn't mean that there are no more messages to read.

There are a few workaround that are not easy or desirable (eg: terminating the topic, or using HTTP admin API to check the backlog on the reader).

Since this is a common theme, we should have a good way to handle this.

Possible solutions

1. When creating a reader, we need a way to tell the current (minimum) amount of messages that it can read. We have that information in the broker, we just need a way to bubble that back to client lib (without breaking API).

2. Provide a way to check if there are messages available to read. Eg.:

   while (reader.hasNext()) {
       Message msg = reader.readNext();
   }

   The tricky part here is that we don't want to ask the broker each time, but rather we should cache that information in client and check periodically. (eg: I check now and it says there are 100 messages, I can check back after those 100 have been read).

3. In some cases, it might desirable to start multiple readers in parallel on the same topic, in a "time-segmented" way.
   For example, if you have 1M message in a single topic, starting 10 readers at different message ids and each of them reading 100K messages in parallel. For this, we would require a way to specify an offset on the reader start message id (eg: messageId + 300K messages).

[sijie]

I am not sure if I understand this correctly. can it achieve by providing a getLastMessageId() call at reader?

so the reader logic can be:

MessageId lastMsgId = reader.getLastMessageId();

Message msg;

while (reader.hasNext()) {
    msg = reader.readNext();

   // process your messages here

   if (msg.getMessageId().compareTo(lastMsgId) == 0) {
        break;
   }
}

[merlimat]

That sounds good as a mechanism under the hood, but I'd rather not allow someone to do things like:

while (true) {
   reader.getLastMessageId();
}

because there would be no flow control at that point :)

[zhaijack]

If it is OK, I would like to contribute on this issue

[ivankelly]

Could we piggyback the last message for the topic on the messages as they are being sent from the broker to the client. The client could then cache this info and if it sees it as being stale enough, make a request to the broker.

[sijie]

This is fixed by #1066

[sijie]

This fix was contributed by @zhaijack