You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2024/06/06 05:45:51 UTC
(pulsar) branch branch-3.3 updated: [improve] [broker] Add additionalSystemCursorNames ignore list for TTL check (#22614)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.3 by this push:
new 79de2daf507 [improve] [broker] Add additionalSystemCursorNames ignore list for TTL check (#22614)
79de2daf507 is described below
commit 79de2daf50700dfb0ecd4fe00ee4af6af775a288
Author: Hang Chen <ch...@apache.org>
AuthorDate: Thu May 9 20:45:56 2024 +0800
[improve] [broker] Add additionalSystemCursorNames ignore list for TTL check (#22614)
---
conf/broker.conf | 4 +
conf/standalone.conf | 4 +
.../apache/pulsar/broker/ServiceConfiguration.java | 7 ++
.../broker/service/persistent/PersistentTopic.java | 7 +-
.../pulsar/broker/service/MessageTTLTest.java | 96 ++++++++++++++++++++++
5 files changed, 117 insertions(+), 1 deletion(-)
diff --git a/conf/broker.conf b/conf/broker.conf
index 1b51ff47551..1ef68a0395c 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 51035235d4d..a8615b70293 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 a9d170ea5de..9efe1856509 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
@@ -652,6 +652,13 @@ public class ServiceConfiguration implements PulsarConfiguration {
)
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<String> 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 7228bdeb2d3..28bc27f7961 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
@@ -40,6 +40,7 @@ import java.util.Map;
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;
@@ -279,6 +280,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
private final ExecutorService orderedExecutor;
private volatile CloseFutures closeFutures;
+ private Set<String> additionalSystemCursorNames = new TreeSet<>();
@Getter
private final PersistentTopicMetrics persistentTopicMetrics = new PersistentTopicMetrics();
@@ -414,6 +416,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
} else {
shadowSourceTopic = null;
}
+ additionalSystemCursorNames = brokerService.pulsar().getConfiguration().getAdditionalSystemCursorNames();
}
@Override
@@ -1934,7 +1937,9 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
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 68a9a769ac1..2f5ad215a1b 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.times;
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 class MessageTTLTest extends BrokerTestBase {
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.setAdditionalSystemCursorNames(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.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<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);
+ }
+
}