[improve] [broker] Add additionalSystemCursorNames ignore list for TTL check

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

@@ -652,6 +652,12 @@ The max allowed delay for delayed delivery (in milliseconds). If the broker rece
         )
     private int ttlDurationDefaultInSeconds = 0;
 
+    @FieldContext(
+        category = CATEGORY_POLICIES,
+        doc = "System subscriptions that will be ignored by ttl check. Default is empty."
+    )
+    private Set<String> systemCursorNames = Set.of("__compaction");
+
     @FieldContext(
         category = CATEGORY_POLICIES,
         dynamic = true,

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

@@ -40,6 +40,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;
@@ -278,6 +279,7 @@ protected TopicStatsHelper initialValue() {
     private final ExecutorService orderedExecutor;
 
     private volatile CloseFutures closeFutures;
+    private Set<String> systemCursorNames = new TreeSet<>();
 
     @Getter
     private final PersistentTopicMetrics persistentTopicMetrics = new PersistentTopicMetrics();
@@ -413,6 +415,7 @@ public PersistentTopic(String topic, ManagedLedger ledger, BrokerService brokerS
         } else {
             shadowSourceTopic = null;
         }
+        systemCursorNames = brokerService.pulsar().getConfiguration().getSystemCursorNames();
     }
 
     @Override
@@ -1933,7 +1936,7 @@ public void checkMessageExpiry() {
         int messageTtlInSeconds = topicPolicies.getMessageTTLInSeconds().get();
         if (messageTtlInSeconds != 0) {
             subscriptions.forEach((__, sub) -> {
-                if (!isCompactionSubscription(sub.getName())) {
+                if (systemCursorNames.contains(sub.getName())) {
                    sub.expireMessages(messageTtlInSeconds);
                 }
             });

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<String> ignoredSubscriptions = new HashSet<>();
+        ignoredSubscriptions.add(subName);
+        int defaultTtl = 5;
+        conf.setSubscriptionsIgnoredByTtl(ignoredSubscriptions);
+        conf.setTtlDurationDefaultInSeconds(defaultTtl);
+        super.baseSetup();
+
+        pulsarClient.newConsumer(Schema.STRING).topic(topicName).subscriptionName(subName)
+                .subscribe().close();
+
+        @Cleanup
+        org.apache.pulsar.client.api.Producer<String> 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> 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.setSubscriptionsIgnoredByTtl(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<String> 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> 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);
+    }
+
 }