[improve][broker] PIP-315: Configurable max delay limit for delayed delivery

conf/broker.conf

@@ -588,6 +588,11 @@ delayedDeliveryMaxNumBuckets=-1
 # fixed delays in messages in a different way.
 delayedDeliveryFixedDelayDetectionLookahead=50000
 
+# The max allowed delay for delayed delivery (in milliseconds). If the broker receives a message which exceeds this
+# max delay, then it will return an error to the producer.
+# The default value is 0 which means there is no limit on the max delivery delay.
+delayedDeliveryMaxDelayInMillis=0
+
 # Whether to enable acknowledge of batch local index.
 acknowledgmentAtBatchIndexLevelEnabled=false
 

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -384,6 +384,12 @@ The delayed message index time step(in seconds) in per bucket snapshot segment,
             + "logic to handle fixed delays in messages in a different way.")
     private long delayedDeliveryFixedDelayDetectionLookahead = 50_000;
 
+    @FieldContext(category = CATEGORY_SERVER, doc = """
+            The max allowed delay for delayed delivery (in milliseconds). If the broker receives a message which \
+            exceeds this max delay, then it will return an error to the producer. \
+            The default value is 0 which means there is no limit on the max delivery delay.""")
+    private long delayedDeliveryMaxDelayInMillis = 0;
+
     @FieldContext(category = CATEGORY_SERVER, doc = "Whether to enable the acknowledge of batch local index")
     private boolean acknowledgmentAtBatchIndexLevelEnabled = false;
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

@@ -515,6 +515,8 @@ protected CompletableFuture<Void> internalSetDelayedDeliveryPolicies(DelayedDeli
                 topicPolicies.setDelayedDeliveryEnabled(deliveryPolicies == null ? null : deliveryPolicies.isActive());
                 topicPolicies.setDelayedDeliveryTickTimeMillis(
                         deliveryPolicies == null ? null : deliveryPolicies.getTickTime());
+                topicPolicies.setDelayedDeliveryMaxDelayInMillis(
+                        deliveryPolicies == null ? null : deliveryPolicies.getMaxDeliveryDelayInMillis());
                 return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies);
             });
     }
@@ -902,6 +904,7 @@ protected CompletableFuture<DelayedDeliveryPolicies> internalGetDelayedDeliveryP
                     delayedDeliveryPolicies = DelayedDeliveryPolicies.builder()
                             .tickTime(policies.getDelayedDeliveryTickTimeMillis())
                             .active(policies.getDelayedDeliveryEnabled())
+                            .maxDeliveryDelayInMillis(policies.getDelayedDeliveryMaxDelayInMillis())
                             .build();
                 }
                 if (delayedDeliveryPolicies == null && applied) {
@@ -910,6 +913,8 @@ protected CompletableFuture<DelayedDeliveryPolicies> internalGetDelayedDeliveryP
                         delayedDeliveryPolicies = DelayedDeliveryPolicies.builder()
                                 .tickTime(pulsar().getConfiguration().getDelayedDeliveryTickTimeMillis())
                                 .active(pulsar().getConfiguration().isDelayedDeliveryEnabled())
+                                .maxDeliveryDelayInMillis(
+                                        pulsar().getConfiguration().getDelayedDeliveryMaxDelayInMillis())
                                 .build();
                     }
                 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java

@@ -242,6 +242,7 @@ protected void updateTopicPolicy(TopicPolicies data) {
         topicPolicies.getReplicatorDispatchRate().updateTopicValue(
             DispatchRateImpl.normalize(data.getReplicatorDispatchRate()));
         topicPolicies.getDelayedDeliveryTickTimeMillis().updateTopicValue(data.getDelayedDeliveryTickTimeMillis());
+        topicPolicies.getDelayedDeliveryMaxDelayInMillis().updateTopicValue(data.getDelayedDeliveryMaxDelayInMillis());
         topicPolicies.getSubscribeRate().updateTopicValue(SubscribeRate.normalize(data.getSubscribeRate()));
         topicPolicies.getSubscriptionDispatchRate().updateTopicValue(
             DispatchRateImpl.normalize(data.getSubscriptionDispatchRate()));
@@ -283,6 +284,9 @@ protected void updateTopicPolicyByNamespacePolicy(Policies namespacePolicies) {
         topicPolicies.getDelayedDeliveryTickTimeMillis().updateNamespaceValue(
                 Optional.ofNullable(namespacePolicies.delayed_delivery_policies)
                         .map(DelayedDeliveryPolicies::getTickTime).orElse(null));
+        topicPolicies.getDelayedDeliveryMaxDelayInMillis().updateNamespaceValue(
+                Optional.ofNullable(namespacePolicies.delayed_delivery_policies)
+                        .map(DelayedDeliveryPolicies::getMaxDeliveryDelayInMillis).orElse(null));
         topicPolicies.getSubscriptionTypesEnabled().updateNamespaceValue(
                 subTypeStringsToEnumSet(namespacePolicies.subscription_types_enabled));
         updateNamespaceReplicatorDispatchRate(namespacePolicies,
@@ -380,6 +384,8 @@ private void updateTopicPolicyByBrokerConfig() {
         topicPolicies.getPublishRate().updateBrokerValue(publishRateInBroker(config));
         topicPolicies.getDelayedDeliveryEnabled().updateBrokerValue(config.isDelayedDeliveryEnabled());
         topicPolicies.getDelayedDeliveryTickTimeMillis().updateBrokerValue(config.getDelayedDeliveryTickTimeMillis());
+        topicPolicies.getDelayedDeliveryMaxDelayInMillis()
+                .updateBrokerValue(config.getDelayedDeliveryMaxDelayInMillis());
         topicPolicies.getCompactionThreshold().updateBrokerValue(config.getBrokerServiceCompactionThresholdInBytes());
         topicPolicies.getReplicationClusters().updateBrokerValue(Collections.emptyList());
         SchemaCompatibilityStrategy schemaCompatibilityStrategy = config.getSchemaCompatibilityStrategy();

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java

@@ -263,6 +263,29 @@ public boolean checkAndStartPublish(long producerId, long sequenceId, ByteBuf he
             }
         }
 
+        if (topic.isPersistent()) {
+            PersistentTopic pTopic = (PersistentTopic) topic;
+            if (pTopic.isDelayedDeliveryEnabled()) {
+                long maxDeliveryDelayInMs = pTopic.getDelayedDeliveryMaxDelayInMillis();
+                if (maxDeliveryDelayInMs > 0) {
+                    headersAndPayload.markReaderIndex();
+                    MessageMetadata msgMetadata = Commands.parseMessageMetadata(headersAndPayload);
+                    headersAndPayload.resetReaderIndex();
+                    if (msgMetadata.hasDeliverAtTime()
+                            && msgMetadata.getDeliverAtTime() - msgMetadata.getPublishTime() > maxDeliveryDelayInMs) {
+                        cnx.execute(() -> {
+                            cnx.getCommandSender().sendSendError(producerId, sequenceId,
+                                    ServerError.NotAllowedError,
+                                    String.format("Exceeds max allowed delivery delay of %s milliseconds",
+                                            maxDeliveryDelayInMs));
+                            cnx.completedSendOperation(false, headersAndPayload.readableBytes());
+                        });
+                        return false;
+                    }
+                }
+            }
+        }
+
         startPublishOperation((int) batchSize, headersAndPayload.readableBytes());
         return true;
     }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

@@ -3724,6 +3724,10 @@ public boolean isDelayedDeliveryEnabled() {
         return topicPolicies.getDelayedDeliveryEnabled().get();
     }
 
+    public long getDelayedDeliveryMaxDelayInMillis() {
+        return topicPolicies.getDelayedDeliveryMaxDelayInMillis().get();
+    }
+
     public int getMaxUnackedMessagesOnSubscription() {
         return topicPolicies.getMaxUnackedMessagesOnSubscription().get();
     }

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiDelayedDeliveryTest.java

@@ -66,6 +66,7 @@ public void testDisableDelayedDelivery() throws Exception {
         DelayedDeliveryPolicies delayedDeliveryPolicies = DelayedDeliveryPolicies.builder()
                 .tickTime(2000)
                 .active(false)
+                .maxDeliveryDelayInMillis(10_000)
                 .build();
         admin.namespaces().setDelayedDeliveryMessages(namespace, delayedDeliveryPolicies);
         //zk update takes time
@@ -124,6 +125,7 @@ public void testNamespaceDelayedDeliveryPolicyApi() throws Exception {
         DelayedDeliveryPolicies delayedDeliveryPolicies = DelayedDeliveryPolicies.builder()
                 .tickTime(3)
                 .active(true)
+                .maxDeliveryDelayInMillis(5000)
                 .build();
         admin.namespaces().setDelayedDeliveryMessages(namespace, delayedDeliveryPolicies);
         Awaitility.await().untilAsserted(()
@@ -151,30 +153,35 @@ public void testDelayedDeliveryApplied() throws Exception {
                 DelayedDeliveryPolicies.builder()
                         .tickTime(conf.getDelayedDeliveryTickTimeMillis())
                         .active(conf.isDelayedDeliveryEnabled())
+                        .maxDeliveryDelayInMillis(conf.getDelayedDeliveryMaxDelayInMillis())
                         .build();
         assertEquals(admin.topics().getDelayedDeliveryPolicy(topic, true), brokerLevelPolicy);
         //set namespace-level policy
         DelayedDeliveryPolicies namespaceLevelPolicy = DelayedDeliveryPolicies.builder()
                 .tickTime(100)
                 .active(true)
+                .maxDeliveryDelayInMillis(4000)
                 .build();
         admin.namespaces().setDelayedDeliveryMessages(namespace, namespaceLevelPolicy);
         Awaitility.await().untilAsserted(()
                 -> assertNotNull(admin.namespaces().getDelayedDelivery(namespace)));
         DelayedDeliveryPolicies policyFromBroker = admin.topics().getDelayedDeliveryPolicy(topic, true);
         assertEquals(policyFromBroker.getTickTime(), 100);
         assertTrue(policyFromBroker.isActive());
+        assertEquals(policyFromBroker.getMaxDeliveryDelayInMillis(), 4000);
         // set topic-level policy
         DelayedDeliveryPolicies topicLevelPolicy = DelayedDeliveryPolicies.builder()
                 .tickTime(200)
                 .active(true)
+                .maxDeliveryDelayInMillis(5000)
                 .build();
         admin.topics().setDelayedDeliveryPolicy(topic, topicLevelPolicy);
         Awaitility.await().untilAsserted(()
                 -> assertNotNull(admin.topics().getDelayedDeliveryPolicy(topic)));
         policyFromBroker = admin.topics().getDelayedDeliveryPolicy(topic, true);
         assertEquals(policyFromBroker.getTickTime(), 200);
         assertTrue(policyFromBroker.isActive());
+        assertEquals(policyFromBroker.getMaxDeliveryDelayInMillis(), 5000);
         //remove topic-level policy
         admin.topics().removeDelayedDeliveryPolicy(topic);
         Awaitility.await().untilAsserted(()

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java

@@ -39,6 +39,7 @@
 import org.apache.pulsar.broker.web.RestException;
 import org.apache.pulsar.common.naming.NamespaceName;
 import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.DelayedDeliveryPolicies;
 import org.apache.pulsar.common.policies.data.DispatchRate;
 import org.apache.pulsar.common.policies.data.PolicyName;
 import org.apache.pulsar.common.policies.data.PolicyOperation;
@@ -196,4 +197,43 @@ public void testOperationDispatchRate() throws Exception {
                 this.testTenant, this.testNamespace));
         assertTrue(Objects.isNull(dispatchRate));
     }
+
+    @Test
+    public void testOperationDelayedDelivery() throws Exception {
+        boolean isActive = true;
+        long tickTime = 1000;
+        long maxDeliveryDelayInMillis = 5000;
+        // 1. set delayed delivery policy
+        namespaces.setDelayedDeliveryPolicies(this.testTenant, this.testNamespace,
+                DelayedDeliveryPolicies.builder()
+                        .active(isActive)
+                        .tickTime(tickTime)
+                        .maxDeliveryDelayInMillis(maxDeliveryDelayInMillis)
+                        .build());
+
+        // 2. query delayed delivery policy & check
+        DelayedDeliveryPolicies policy =
+                (DelayedDeliveryPolicies) asyncRequests(response -> namespaces.getDelayedDeliveryPolicies(response,
+                        this.testTenant, this.testNamespace));
+        assertEquals(policy.isActive(), isActive);
+        assertEquals(policy.getTickTime(), tickTime);
+        assertEquals(policy.getMaxDeliveryDelayInMillis(), maxDeliveryDelayInMillis);
+
+        // 3. remove & check
+        namespaces.removeDelayedDeliveryPolicies(this.testTenant, this.testNamespace);
+        policy =
+                (DelayedDeliveryPolicies) asyncRequests(response -> namespaces.getDelayedDeliveryPolicies(response,
+                        this.testTenant, this.testNamespace));
+        assertTrue(Objects.isNull(policy));
+
+        // 4. invalid namespace check
+        String invalidNamespace = this.testNamespace + "/";
+        try {
+            namespaces.setDelayedDeliveryPolicies(this.testTenant, invalidNamespace,
+                    DelayedDeliveryPolicies.builder().build());
+            fail("should have failed");
+        } catch (RestException e) {
+            assertEquals(e.getResponse().getStatus(), Response.Status.PRECONDITION_FAILED.getStatusCode());
+        }
+    }
 }

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java

@@ -71,6 +71,7 @@
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.policies.data.BacklogQuota;
 import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.DelayedDeliveryPolicies;
 import org.apache.pulsar.common.policies.data.DispatchRate;
 import org.apache.pulsar.common.policies.data.HierarchyTopicPolicies;
 import org.apache.pulsar.common.policies.data.InactiveTopicDeleteMode;
@@ -3157,4 +3158,28 @@ public void testProduceChangesWithEncryptionRequired() throws Exception {
                 });
     }
 
+    @Test
+    public void testDelayedDeliveryPolicy() throws Exception {
+        final String topic = testTopic + UUID.randomUUID();
+        admin.topics().createNonPartitionedTopic(topic);
+
+        boolean isActive = true;
+        long tickTime = 1000;
+        long maxDeliveryDelayInMillis = 5000;
+        DelayedDeliveryPolicies policy = DelayedDeliveryPolicies.builder()
+                .active(isActive)
+                .tickTime(tickTime)
+                .maxDeliveryDelayInMillis(maxDeliveryDelayInMillis)
+                .build();
+
+        admin.topicPolicies().setDelayedDeliveryPolicy(topic, policy);
+        Awaitility.await()
+                .untilAsserted(() -> Assert.assertEquals(admin.topicPolicies().getDelayedDeliveryPolicy(topic), policy));
+
+        admin.topicPolicies().removeDelayedDeliveryPolicy(topic);
+        Awaitility.await()
+                .untilAsserted(() -> Assert.assertNull(admin.topicPolicies().getDelayedDeliveryPolicy(topic)));
+
+        admin.topics().delete(topic, true);
+    }
 }

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/DelayedDeliveryTest.java

@@ -23,6 +23,7 @@
 import static org.testng.Assert.assertNotNull;
 import static org.testng.Assert.assertNull;
 import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
@@ -337,6 +338,7 @@ public void testEnableAndDisableTopicDelayedDelivery() throws Exception {
         DelayedDeliveryPolicies delayedDeliveryPolicies = DelayedDeliveryPolicies.builder()
                 .tickTime(2000)
                 .active(false)
+                .maxDeliveryDelayInMillis(5000)
                 .build();
         admin.topics().setDelayedDeliveryPolicy(topicName, delayedDeliveryPolicies);
         //wait for update
@@ -349,6 +351,7 @@ public void testEnableAndDisableTopicDelayedDelivery() throws Exception {
 
         assertFalse(admin.topics().getDelayedDeliveryPolicy(topicName).isActive());
         assertEquals(2000, admin.topics().getDelayedDeliveryPolicy(topicName).getTickTime());
+        assertEquals(5000, admin.topics().getDelayedDeliveryPolicy(topicName).getMaxDeliveryDelayInMillis());
 
         admin.topics().removeDelayedDeliveryPolicy(topicName);
         //wait for update
@@ -622,4 +625,42 @@ public void testDispatcherReadFailure() throws Exception {
         }
     }
 
+    @Test
+    public void testDelayedDeliveryExceedsMaxDelay() throws Exception {
+        long maxDeliveryDelayInMillis = 5000;
+        String topic = BrokerTestUtil.newUniqueName("testDelayedDeliveryExceedsMaxDelay");
+
+        @Cleanup
+        Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+                .topic(topic)
+                .create();
+
+        admin.topicPolicies().setDelayedDeliveryPolicy(topic,
+                DelayedDeliveryPolicies.builder()
+                        .active(true)
+                        .tickTime(100L)
+                        .maxDeliveryDelayInMillis(maxDeliveryDelayInMillis)
+                        .build());
+
+        //wait for update
+        for (int i = 0; i < 50; i++) {
+            Thread.sleep(100);
+            if (admin.topics().getDelayedDeliveryPolicy(topic) != null) {
+                break;
+            }
+        }
+
+        try {
+            producer.newMessage()
+                    .value("msg")
+                    .deliverAfter(6, TimeUnit.SECONDS)
+                    .send();
+
+            producer.flush();
+            fail("Should have thrown NotAllowedException");
+        } catch (PulsarClientException.NotAllowedException ex) {
+            assertEquals(ex.getMessage(), "Exceeds max allowed delivery delay of "
+                    + maxDeliveryDelayInMillis + " milliseconds");
+        }
+    }
 }

pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/DelayedDeliveryPolicies.java

@@ -26,10 +26,12 @@
 public interface DelayedDeliveryPolicies {
     long getTickTime();
     boolean isActive();
+    long getMaxDeliveryDelayInMillis();
 
     interface Builder {
         Builder tickTime(long tickTime);
         Builder active(boolean active);
+        Builder maxDeliveryDelayInMillis(long maxDeliveryDelayInMillis);
         DelayedDeliveryPolicies build();
     }
 

pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/impl/DelayedDeliveryPoliciesImpl.java

@@ -32,6 +32,7 @@
 public final class DelayedDeliveryPoliciesImpl implements DelayedDeliveryPolicies {
     private long tickTime;
     private boolean active;
+    private long maxDeliveryDelayInMillis;
 
     public static DelayedDeliveryPoliciesImplBuilder builder() {
         return new DelayedDeliveryPoliciesImplBuilder();
@@ -40,6 +41,7 @@ public static DelayedDeliveryPoliciesImplBuilder builder() {
     public static class DelayedDeliveryPoliciesImplBuilder implements DelayedDeliveryPolicies.Builder {
         private long tickTime;
         private boolean active;
+        private long maxDeliveryDelayInMillis;
 
         public DelayedDeliveryPoliciesImplBuilder tickTime(long tickTime) {
             this.tickTime = tickTime;
@@ -51,8 +53,13 @@ public DelayedDeliveryPoliciesImplBuilder active(boolean active) {
             return this;
         }
 
+        public DelayedDeliveryPoliciesImplBuilder maxDeliveryDelayInMillis(long maxDeliveryDelayInMillis) {
+            this.maxDeliveryDelayInMillis = maxDeliveryDelayInMillis;
+            return this;
+        }
+
         public DelayedDeliveryPoliciesImpl build() {
-            return new DelayedDeliveryPoliciesImpl(tickTime, active);
+            return new DelayedDeliveryPoliciesImpl(tickTime, active, maxDeliveryDelayInMillis);
         }
     }
 }