Deferred messages for consumers

[lovelle]

Motivation

Normally messages are received by the consumer as soon as they were published by
producer. This feature offers the capability to configure consumer subscription
with an arbitrary receiver delay.
Receiver delay means that consumers will only receive messages that are older
than this parameter plus publish time as well. Messages that are not ready to be
delivered are schedule for deferred delivery.

This is helpful on systems relaying on the producer/consumer pattern used for
synchronisation or healthy checks, but on such systems is common to have some
overhead committing data on persistent storage maybe due to buffered mechanism
or distributing the data across the network before being available,
e.g. Elasticsearch

With this feature users will be able to do these kind of work with ease and
without doing any ugly hack like using re-delivery and testing until any
condition is met after some time.

Modifications

pulsar-broker

• Add field for DelayQueue to store next positions pending of delivery.

• Add field for ScheduledFuture to schedule the head of DelayQueue in order to
  start the process of delivering expired entries.

• Fix on sendMessages method usage of received list of entries when such
  entries are being filtered by updatePermitsAndPendingAcks method.
  Received list of entries might be filtered by updatePermitsAndPendingAcks
  method and afterwards entries are processed by execute() with lambda which
  is probably that it's executing thread will be other than its calling thread.

  Therefore entries are now wrapped with CopyOnWriteArrayList in order
  to prevent ConcurrentModificationException by lambda on execute(), another
  approach could be to copy the entire list for lambda or to use a synchronized
  list, but this would result in a performance penalty even when entries are
  not being filtered, CopyOnWriteArrayList prevent this from happening.

  Another step further trying to fix this might be using Streams and applying
  transformations to inner list. This path was not taken because would
  require major changes.

• Add processDelayEntries() method to process all elements added in DelayQueue
  which are ready to be delivered, at any given time just one task should be
  schedule using this method.

• Add readPublishFrom() method to get the parameter of publish time from
  metadata of a message without changing its reference offset, this method
  will only be used if the consumer has enabled the receiver delay parameter
  on the subscription.

• Add inner private class DelayPositionInfo to represent each position to be
  schedule in DelayQueue.

• Add method to clean-up previous mentioned fields related to deferred
  messages.

DelayQueue from java.util.concurrent is used in order to store each messages
position next to be expired, the advantage of using this queue is that at any
given time one and only one task is scheduled per consumer avoiding to schedule
an unbounded number of tasks.

pulsar-client

• Add receiverDelay() method at subscription level to configure this parameter.

pulsar-common

• Set receiver delay whether it was configured by user on subscription.

• Add optional receiver delay parameter to protobuf pulsar schema,
  code generated by generate_protobuf_docker.sh script.

Result

For those consumers that have been configured with a receiver delay, broker
service will be able to deliver only messages that at some given moment in time
are ready to deliver.
Ready to delivery means: now() < publishTime() + receiverDelay

Tests

I've just tested this feature with an external application consuming messages
at different receiver delay rates.

If this feature is accepted I will be pleased to add all needed tests and also
to add receiver delay on all supported languages.

Future considerations

An improvement could be made with the way each position is stored on DelayQueue,
instead of storing the position of each message, when messages are being
receiving sequentially (this should be the common case) a range of positions
could be stored on a single element on DelayQueue saving memory usage.

This improvement was not made in order to leave this feature as simple as
possible, if this feature is accepted I would like to do this if pulsar
community agrees.

[sijie]

@codelipenghui this seems to be related to #2375 . can you review it?

[codelipenghui]

If support fixed delays on all message in a topic. I recommend implement on the client side to keep server-side message dispatch simple. Pulsar client already expose redelivery count every message. We can add some control to the re-delivery message on client-side.

[lovelle]

Hi! @sijie I think #2375 is a bit different from this because is proposed to be 'publisher dependent' being the publisher the one who decides which delay will have each message instead of the consumer, besides that, in the way this feature is implemented I think it would be easy to support #2375 with the current implementation from this pr.

@codelipenghui I also thought in doing this at client side but I see two major downsides about it:

1. There would be a lot of I/O overhead wasted between clients and brokers delivering unnecessary messages.
2. The decision making of expired messages will be made at client side which depends on client clock, being more error prone and difficult to detect and fix by an administrator.

[codelipenghui]

@lovelle

The decision making of expired messages will be made at client side which depends on client clock, being more error prone and difficult to detect and fix by an administrator.

If can't depends on client clock, i think publish time now be client-generated

There would be a lot of I/O overhead wasted between clients and brokers delivering unnecessary messages.

Your worry is very necessary.

Through our discussion, we can get the following conclusions:

1. Deferred messages consume can't depends on client clock include message publish time.
2. Implement deferred message consume on client-side has lot of downsides(I/O waste, depends on client clock, un-acked messages limitation...)

If on server-side, can delayed dispatch be simpler?

Dispatcher read entries from ledger, i think we just need to check tail message(if generated on server side, publish time is already sorted)

[lovelle]

@codelipenghui

If can't depends on client clock, i think publish time now be client-generated

Yes, as far as I understand publish time is set here this should be set by broker instead of client, do you see any problem in doing so?

Just to clarify this one step further, I can see two main features with the delayed message topic.

1. Consumer dependent delay (this pr) would be used just to having some delay on every message of the consumer, maybe for short period of times, in the range or less than minutes. And the obvious advantage is that the producer does not need to be aware of such delay.

2. Publisher dependent delay (Support for delayed message delivery #2375 and here) I think this feature is more advance and complex and the proposed way is a really good path, since the time of delay for deliver such messages could be far in time and totally different from each other, therefore, messages need to be structured in consideration with this factor.

I think both approaches would be great to have on pulsar since they are used for different purposes.

So, on 2nd approach we could do what you propose if I'm not wrong.

[codelipenghui]

@lovelle

You can view the discussion about PIP-26 on the mailing list.

@jiazhai @sijie Can you give some advice? I tend to implement consumer dependent delay on server-side by a simple way.

[sijie]

@lovelle @codelipenghui

I took a look at this PR. I agreed with Consumer dependent delay (this pr) is a very useful feature. Since its goal is basically delaying delivering messages for a given configured interval, I think it is easy to add this feature at broker side. because if you implement this at broker side, it is available across different language clients. Currently implementation only works for java client. If we want this feature to be available at c/c++/python/go, we have to reimplement it at c++ client, which I think it is non-trivial.

@lovelle I am also open to see what is your opinion about implementing this at broker side vs at client side.

[lovelle]

@sjie

I totally agree with you! I think this feature is better suited at broker side, first reason as you said, would be simplicity and much less code on the overall pulsar project. Second reason is clock dependent at back end side instead at client side, giving more flexibility to an administrator. Third reason would be to save a lot of I/O wasted for delivering messages not expired.

This pr as is, the only thing that adds at client side is the ability to pass an optional parameter (receiver delay) to the broker, which is a period of time to let the broker do the calculation of which messages are expired.
If all of you agrees to go on with this pr I'm able to add this parameter to the rest of the clients, maybe on a dedicated pr?

In my opinion there is one thing we need to solve before going on with this pr, as @codelipenghui said, publish time is client-generated, we need to change this in order to set this parameter at broker level to use it in a safer way. Do you see any problem doing this change?

[sijie]

If all of you agrees to go on with this pr I'm able to add this parameter to the rest of the clients, maybe on a dedicated pr?

+1 from me

as @codelipenghui said, publish time is client-generated, we need to change this in order to set this parameter at broker level to use it in a safer way. Do you see any problem doing this change?

that seems to be okay to me. however for backward compatability consideration, I would suggest adding a flag on producer to control set publisher time or not at client side. If producer doesn't set publish time, the broker will then set the publish time.

[lovelle]

@sijie

I've just created #3267 in a separated pr, so we can treat them independently and discuss each change without making things more fuzzy.

[sijie]

@lovelle sounds good. thank you!

[sijie]

Closed this pull request in favor of the implementation in #4062