diff --git a/conf/broker.conf b/conf/broker.conf index e0ebbe3043aad..c5beda206a4db 100644 --- a/conf/broker.conf +++ b/conf/broker.conf @@ -180,6 +180,10 @@ backlogQuotaDefaultRetentionPolicy=producer_request_hold # Default ttl for namespaces if ttl is not already configured at namespace policies. (disable default-ttl with value 0) ttlDurationDefaultInSeconds=0 +# Additional system subscriptions that will be ignored by ttl check. The cursor names are comma separated. +# Default is empty. +# additionalSystemCursorNames= + # Enable topic auto creation if new producer or consumer connected (disable auto creation with value false) allowAutoTopicCreation=true diff --git a/conf/standalone.conf b/conf/standalone.conf index 5eb9fadcf19dc..5ca0c683d7438 100644 --- a/conf/standalone.conf +++ b/conf/standalone.conf @@ -121,6 +121,10 @@ backlogQuotaDefaultLimitSecond=-1 # Default ttl for namespaces if ttl is not already configured at namespace policies. (disable default-ttl with value 0) ttlDurationDefaultInSeconds=0 +# Additional system subscriptions that will be ignored by ttl check. The cursor names are comma separated. +# Default is empty. +# additionalSystemCursorNames= + # Enable the deletion of inactive topics. This parameter need to cooperate with the allowAutoTopicCreation parameter. # If brokerDeleteInactiveTopicsEnabled is set to true, we should ensure that allowAutoTopicCreation is also set to true. brokerDeleteInactiveTopicsEnabled=true diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java index f53eb7e183f5c..10ffced0321de 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java @@ -646,6 +646,13 @@ The delayed message index time step(in seconds) in per bucket snapshot segment, ) private int ttlDurationDefaultInSeconds = 0; + @FieldContext( + category = CATEGORY_POLICIES, + doc = "Additional system subscriptions that will be ignored by ttl check. " + + "The cursor names are comma separated. Default is empty." + ) + private Set additionalSystemCursorNames = new TreeSet<>(); + @FieldContext( category = CATEGORY_POLICIES, dynamic = true, diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index b4f295dbd973f..472387a0a9bce 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -39,6 +39,7 @@ import java.util.Objects; import java.util.Optional; import java.util.Set; +import java.util.TreeSet; import java.util.concurrent.CancellationException; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; @@ -272,6 +273,7 @@ protected TopicStatsHelper initialValue() { private final ExecutorService orderedExecutor; private volatile CloseFutures closeFutures; + private Set additionalSystemCursorNames = new TreeSet<>(); /*** * We use 3 futures to prevent a new closing if there is an in-progress deletion or closing. We make Pulsar return @@ -384,6 +386,7 @@ public PersistentTopic(String topic, ManagedLedger ledger, BrokerService brokerS } else { shadowSourceTopic = null; } + additionalSystemCursorNames = brokerService.pulsar().getConfiguration().getAdditionalSystemCursorNames(); } @Override @@ -1888,7 +1891,9 @@ public void checkMessageExpiry() { int messageTtlInSeconds = topicPolicies.getMessageTTLInSeconds().get(); if (messageTtlInSeconds != 0) { subscriptions.forEach((__, sub) -> { - if (!isCompactionSubscription(sub.getName())) { + if (!isCompactionSubscription(sub.getName()) + && (additionalSystemCursorNames.isEmpty() + || !additionalSystemCursorNames.contains(sub.getName()))) { sub.expireMessages(messageTtlInSeconds); } }); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessageTTLTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessageTTLTest.java index 68a9a769ac1fe..2f5ad215a1b6e 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessageTTLTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessageTTLTest.java @@ -23,15 +23,23 @@ import static org.mockito.Mockito.verify; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertTrue; + import java.util.ArrayList; +import java.util.HashSet; import java.util.List; +import java.util.Optional; +import java.util.Set; +import java.util.TreeSet; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import lombok.Cleanup; import org.apache.bookkeeper.mledger.impl.PositionImpl; +import org.apache.pulsar.broker.service.persistent.PersistentSubscription; import org.apache.pulsar.broker.service.persistent.PersistentTopic; import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.impl.MessageIdImpl; import org.apache.pulsar.common.policies.data.ManagedLedgerInternalStats.CursorStats; import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats; @@ -138,4 +146,92 @@ public void testTTLPoliciesUpdate() throws Exception { topicRefMock.onUpdate(topicPolicies); verify(topicRefMock, times(2)).checkMessageExpiry(); } + + @Test + public void testTtlFilteredByIgnoreSubscriptions() throws Exception { + String topicName = "persistent://prop/ns-abc/testTTLFilteredByIgnoreSubscriptions"; + String subName = "__SUB_FILTER"; + cleanup(); + Set ignoredSubscriptions = new HashSet<>(); + ignoredSubscriptions.add(subName); + int defaultTtl = 5; + conf.setAdditionalSystemCursorNames(ignoredSubscriptions); + conf.setTtlDurationDefaultInSeconds(defaultTtl); + super.baseSetup(); + + pulsarClient.newConsumer(Schema.STRING).topic(topicName).subscriptionName(subName) + .subscribe().close(); + + @Cleanup + org.apache.pulsar.client.api.Producer producer = pulsarClient.newProducer(Schema.STRING) + .enableBatching(false).topic(topicName).create(); + + final int messages = 10; + + for (int i = 0; i < messages; i++) { + String message = "my-message-" + i; + producer.send(message); + } + producer.close(); + + Optional topic = pulsar.getBrokerService().getTopicReference(topicName); + assertTrue(topic.isPresent()); + PersistentSubscription subscription = (PersistentSubscription) topic.get().getSubscription(subName); + + Thread.sleep((defaultTtl - 1) * 1000); + topic.get().checkMessageExpiry(); + // Wait the message expire task done and make sure the message does not expire early. + Thread.sleep(1000); + assertEquals(subscription.getNumberOfEntriesInBacklog(false), 10); + Thread.sleep(2000); + topic.get().checkMessageExpiry(); + // Wait the message expire task done. + retryStrategically((test) -> subscription.getNumberOfEntriesInBacklog(false) == 0, 5, 200); + // The message should not expire because the subscription is ignored. + assertEquals(subscription.getNumberOfEntriesInBacklog(false), 10); + + conf.setAdditionalSystemCursorNames(new TreeSet<>()); + } + + @Test + public void testTtlWithoutIgnoreSubscriptions() throws Exception { + String topicName = "persistent://prop/ns-abc/testTTLWithoutIgnoreSubscriptions"; + String subName = "__SUB_FILTER"; + cleanup(); + int defaultTtl = 5; + conf.setTtlDurationDefaultInSeconds(defaultTtl); + conf.setBrokerDeleteInactiveTopicsEnabled(false); + super.baseSetup(); + + pulsarClient.newConsumer(Schema.STRING).topic(topicName).subscriptionName(subName) + .subscribe().close(); + + @Cleanup + org.apache.pulsar.client.api.Producer producer = pulsarClient.newProducer(Schema.STRING) + .enableBatching(false).topic(topicName).create(); + + final int messages = 10; + + for (int i = 0; i < messages; i++) { + String message = "my-message-" + i; + producer.send(message); + } + producer.close(); + + Optional topic = pulsar.getBrokerService().getTopicReference(topicName); + assertTrue(topic.isPresent()); + PersistentSubscription subscription = (PersistentSubscription) topic.get().getSubscription(subName); + + Thread.sleep((defaultTtl - 1) * 1000); + topic.get().checkMessageExpiry(); + // Wait the message expire task done and make sure the message does not expire early. + Thread.sleep(1000); + assertEquals(subscription.getNumberOfEntriesInBacklog(false), 10); + Thread.sleep(2000); + topic.get().checkMessageExpiry(); + // Wait the message expire task done and make sure the message expired. + retryStrategically((test) -> subscription.getNumberOfEntriesInBacklog(false) == 0, 5, 200); + assertEquals(subscription.getNumberOfEntriesInBacklog(false), 0); + } + }