315157973 commented on code in PR #21490:
URL: https://github.com/apache/pulsar/pull/21490#discussion_r1386296475
##########
pip/pip_315.md:
##########
@@ -0,0 +1,142 @@
+# PIP-315: Configurable max delay limit for delayed delivery
+
+# Background knowledge
+Delayed message delivery is an important feature which allows a producer to
specify that a message should be delivered/consumed at a later time. Currently
the broker will save a delayed message without any check. The message's
`deliverAt` time is checked when the broker dispatches messages to the
Consumer. If a message has a `deliverAt` time, then it is added to the
`DelayedDeliveryTracker` and will be delivered later when eligible.
+
+Delayed message delivery is only available for persistent topics, and
shared/key-shared subscription types.
+
+# Motivation
+Currently there is no max delay limit so a producer can specify any delay when
publishing a message.
+
+This poses a few challenges:
+1. Producer may miscalculate/misconfigure a very large delay (ex. 1,000 day
instead of 100 day delay)
+2. Pulsar administrators may want to limit the max allowed delay since unacked
messages (ex. messages with a large delay) will be stored forever (unless TTL
is configured)
+3. The configured delay may be greater than the configured TTL which means the
delayed message may be deleted before the `deliverAt` time (before the consumer
can process it)
+
+# Goals
+The purpose of this PIP is to introduce an optional configuration to limit the
max allowed delay for delayed delivery.
+
+## In Scope
+- Add broker configuration to limit the max allowed delay for delayed delivery
+- Configurable at broker/topic/namespace-level
+
+## Out of Scope
+
+# High Level Design
+We will add a configuration `maxDeliveryDelayInSeconds` and if configured, the
broker will check incoming delayed messages to see if the message's `deliverAt`
time exceeds the configured limit. If it exceeds the limit, the broker will
send an error back to the Producer.
+
+# Detailed Design
+
+## Design & Implementation Details
+
+### Broker Changes
+A new `maxDeliveryDelayInSeconds` config will be added to the broker which is
initially defaulted to 0 (disabled). The default (disabled) behavior will match
the current delayed delivery behavior (no limit on delivery delay).
+```
+# broker.conf
+delayedDeliveryMaxDeliveryDelayInSeconds=0
+```
+
+This field will also be added to the existing `DelayedDeliveryPolicies`
interface to support topic & namespace-level configuration:
+```java
+public interface DelayedDeliveryPolicies {
+ long getMaxDeliveryDelayInSeconds();
+}
+```
+
+The max delivery delay check will occur in the broker's `Producer` class
inside of `checkAndStartPublish` (same place as other checks such as
`isEncryptionEnabled`).
+
+We will give a `ServerError.NotAllowedError` error if all of the following are
true:
+1. Sending to a persistent topic
+2. Topic has `delayedDeliveryEnabled=true`
+3. `MessageMetadata` `deliver_at_time` has been specified
+4. Topic has `>0` value for `maxDeliveryDelayInSeconds`
+5. `deliver_at_time` > `System.currentTimeMillis() + maxDeliveryDelayInSeconds
* 1000`
+
+```java
+// In org.apache.pulsar.broker.service.Producer#checkAndStartPublish
+if (topic.isPersistent()) {
+ PersistentTopic pTopic = (PersistentTopic) topic;
+ if (pTopic.isDelayedDeliveryEnabled()) {
+ headersAndPayload.markReaderIndex();
+ MessageMetadata msgMetadata =
Commands.parseMessageMetadata(headersAndPayload);
+ headersAndPayload.resetReaderIndex();
+ if (msgMetadata.hasDeliverAtTime()) {
+ long deliverAtTime = msgMetadata.getDeliverAtTime();
+ long maxDeliveryDelayInSeconds =
pTopic.getMaxDeliveryDelayInSeconds();
+ if (maxDeliveryDelayInSeconds > 0
+ && deliverAtTime > System.currentTimeMillis() +
maxDeliveryDelayInSeconds * 1000) {
Review Comment:
This seems like a good proposal. I have some comments
1. Now the delay time accuracy reaches millisecond level, and the unit of
the maximum delay threshold should also be consistent.
2. Each message will be judged, so we should try to reduce unnecessary
repeated calculations, such as `maxDeliveryDelayInSeconds` * 1000
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
