I am not a fan of adding complexity to the dispatch path, and I will always
have serious concerns about proposals that do so, including this one.
In general, I would prefer Pulsar to keep the dispatch path simple and
efficient, and avoid server side implementations of business logic.
Streaming at scale, at low latency is what I think Pulsar should do. I am
biased here, because that is one of the reasons Pulsar got created
originally, at a time when there were many other message brokers out there
( and many under the Apache umbrella too)
All those other message brokers do all kinds of server-side logic -
filtering, transforming, scheduling, and so on. All of those systems have
more or less ended up with bottlenecks and complexity. And this is not
without reason. Message queues are queues, and most of the server side
logic implementations are attempts to make a queue into a database. A
system that is optimized for flow as a queue, will not be good as a
database, and vice-versa.
I think the right way to do this kind of business logic is in the client or
leverage Pulsar functions, and the core broker dispatch path and process
space should just deal with performance and flow at scale
Joe
On Thu, Nov 8, 2018 at 1:39 PM 李鹏辉gmail <codelipeng...@gmail.com> wrote:
> Dear all
>
> This is a PIP to add feature of delayed message delivery.
>
> ## Motivation
> Scheduled and delayed message delivery is a very common feature to support
> in a message system. Basically individual message can have a header which
> will be set by publisher and based on the header value the broker should
> hold on delivering messages until the configured delay or the scheduled
> time is met.
>
> ## Usage
> The delayed message delivery feature is enabled per message at producer
> side.
>
> Delayed messages publish example in client side:
>
> ```java
> // message to be delivered at the configured delay interval
> producer.newMessage().delayAt(3L, TimeUnit.Minute).value("Hello
> Pulsar!").send();
>
> // message to be delivered at the configure time.
> producer.newMessage().scheduleAt(new Date(2018, 10, 31, 23, 00, 00))
> ```
>
> To enable or disable delay message feature:
>
> ```shell
> pulsar-admin namespaces
>
> enable-delayed-message Enable delayed message for all topics of the
> namespace
> Usage: enable-delayed-message [options] tenant/namespace
>
> Options:
> -p --time-partition-granularity
> Granularities of time will be partitioned, every time
> partition will be
> stored into legders and current time partition will be
> load in memory
> and organized in a TimeWheel.(eg: 30s, 5m, 1h, 3d, 2w)
> Default: 5m
> -t --tick-duration
> The duration between tick in TimeWheel. Calculate ticks
> per wheel
> using time-partition-granularity / tick-duration before
> load time
> partition into a TimeWheel.(eg: 500ms, 1s, 5m)
> Default: 1s
>
> disable-delayed-message Disable delayed message for all topics of
> the namespace
> Usage: disable-delayed-message tenant/namespace
> ```
>
> ## Design
>
> ### Delayed Message Index
>
> The “DelayedMessageIndex” will be implemented using a [TimeWheel approach](
> http://www.cs.columbia.edu/~nahum/w6998/papers/sosp87-timing-wheels.pdf).
> We will be maintaining a delayed index, indexing the delayed message by its
> time and actual message id.
>
> The index is partitioned by the delayed time. Each time partition will be
> stored using one (or few) ledger(s). For example, if we are configuring
> the index to be partitioned by 5 minutes, we will store the index data for
> every 5 minutes by its delayed time. The latest time partition will be
> loaded in memory and organized in a TimeWheel.
>
> The TimeWheel is indexed by ticks. For example, if we configured the tick
> to be 1 second, we will be maintaining 300 ticks for 5 minutes’ index. A
> timer task is scheduled every tick, and it will pick the indexed message
> from the TimeWheel and dispatch them to the real consumers.
>
> After completing dispatching the messages in current TimeWheel, it will
> load the TimeWheel from the next time partition.
>
> Delayed message option ` time-partition-granularity ` and `tick-duration`
> properly be reset to adapt delay message throughput change. `
> time-partition-granularity ` can't be shrink. For example, exist config is
> time-partition-granularity = 5m and tick-duration = 1s, delay message index
> will store in 300 slot, If increase the time-partition-granularity to 10m,
> when load next time partition TimeWheel will init with 600 slot, Timewheel
> has enough slot to maintain already exist time partition(5m), but if
> decrease the time-partition-granularity to 2m, Timewheel can't load already
> exist time partition(5m) into 120 slot. So regardless the
> time-partition-granularity shrink first, It's can be improve by split time
> partition when load time partition. For time-partition-granularity,
> increase or decrease just affect precision of delay time.
>
> Delay message feature conflict with TTL and Backlog
> Quota(consumer_backlog_eviction). So it's necessary write a doc to explain
> the conflict result. Next, if support ledger compact feature, it's can be
> improve by this feature, relocate un-ack messages to a new ledger, old
> ledger can be safe delete.
>
> ### Delayed Message Index Cursor
>
> We will maintain a `cursor` for ensure the messages will be indexed
> correctly in the `DelayMessageIndex`, so we don’t miss dispatching any
> messages.
>
> ### Recovery
>
> Since we are organizing the delayed message index using ledgers, and have
> a cursor to ensure messages are correctly indexed. So when a broker fails
> due to any reason, and the topic is recovered, we can figure out the latest
> time partition and load the index from the ledgers.
>
> ## Changes
>
> ### Protocol Changes
>
> In order to support `delayed` and `scheduled` messages, we need to add
> following fields in `MessageMetadata`.
>
> ```protobuf
> message MessageMetadata {
>
> // the message will be delayed at delivery by `delayed_ms`
> milliseconds.
> optional int64 delayed_ms = 18;
>
> }
> ```
>
> ### Broker Changes
>
> The publish path will not be changed. All the messages will still be first
> appended to the managed ledger as normal. As the delivery related message
> attributes will be ignored for `failover` and `exclusive` subscriptions to
> ensure FIFO ordering.
>
> The change happens at `MultipleConsumers` dispatcher. At the dispatching
> time, the dispatcher will check the delivery attributes at the message
> header. If a message is qualified for delivery (exceeding the configured
> delayed/scheduled time), the message will be dispatched as normal;
> otherwise it will be added back to the “delayed message index”. Those
> delayed messages will be delayed at dispatching from the “delayed message
> index”.
>
> ### Client Changes
>
> #### TypedMessageBuilder
>
> ```java
> public interface TypedMessageBuilder<T> {
>
> /**
> * Build a message to be delivered at the configured delay interval :
> <tt>delayTime</tt>.
> * @return typed message builder.
> */
> TypedMessageBuilder<T> delayedAt(long delayTime, TimeUnit
> timeUnit);
>
> /**
> * Build a message to be delivered at the configure time.
> *
> * @return typed message builder.
> */
> TypedMessageBuilder<T> scheduledAt(Date date);
> }
> ```
>
> #### Producer
>
> Delay message can not send in bulk, because define the delay property in
> MessageMetadata.
>
> #### Consumer
>
> Both the `delayed` and `scheduled` messages will be appended to the topic.
> The `delayed` and `scheduled` message attributes are only applied to
> `shared` subscription. It means both exclusive and failover subscriptions
> will be ignoring these message delivery attributes to ensure FIFO delivery.
>
> ## Test Plan
>
> Unit tests + Integration Tests
>
>
>
> https://docs.google.com/document/d/1yZYPJR6ZO9mJdeMgzbi4MJlp_KmKnU2cjbJmmDYyDI0/edit?usp=sharing
>
>
> —
> Regards,
> Penghui Li
