You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/07/22 12:05:51 UTC
(pulsar) 11/16: [improve][broker] Skip system cursor when check inactive cursor. (#26149)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit f7591e3a86701c6d6a0f112aeb19900809c301e7
Author: Yan Zhao <ho...@apache.org>
AuthorDate: Thu Jul 16 11:53:05 2026 +0800
[improve][broker] Skip system cursor when check inactive cursor. (#26149)
(cherry picked from commit fe61afda2d5bf4cb2de019cca47ef78cd1e350f1)
---
.../apache/pulsar/broker/service/persistent/PersistentTopic.java | 2 +-
.../java/org/apache/pulsar/broker/service/PersistentTopicTest.java | 7 +++++++
2 files changed, 8 insertions(+), 1 deletion(-)
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 558d51b8e050..798c5cc60697 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
@@ -3560,7 +3560,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
subscriptions.forEach((subName, sub) -> {
if (sub.dispatcher != null && sub.dispatcher.isConsumerConnected()
|| sub.isReplicated()
- || isCompactionSubscription(subName)) {
+ || isSystemCursor(subName)) {
return;
}
if (System.currentTimeMillis() - sub.cursor.getLastActive() > expirationTimeMillis) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
index 011b47464ebf..4872fbd2f0d1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
@@ -2027,6 +2027,7 @@ public class PersistentTopicTest extends MockedBookKeeperTestCase {
@Test
public void testCheckInactiveSubscriptions() throws Exception {
+ pulsarTestContext.getConfig().setAdditionalSystemCursorNames(Set.of("additionalSystemCursor"));
PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService);
final var subscriptions = new ConcurrentHashMap<String, PersistentSubscription>();
@@ -2045,6 +2046,11 @@ public class PersistentTopicTest extends MockedBookKeeperTestCase {
spyWithClassAndConstructorArgsRecordingInvocations(PersistentSubscription.class, topic,
"nonDeletableSubscription2", cursorMock, true);
subscriptions.put(nonDeletableSubscription2.getName(), nonDeletableSubscription2);
+ // This subscription is an additional system cursor.
+ PersistentSubscription nonDeletableSubscription3 =
+ spyWithClassAndConstructorArgsRecordingInvocations(PersistentSubscription.class, topic,
+ "additionalSystemCursor", cursorMock, false);
+ subscriptions.put(nonDeletableSubscription3.getName(), nonDeletableSubscription3);
Field field = topic.getClass().getDeclaredField("subscriptions");
field.setAccessible(true);
@@ -2071,6 +2077,7 @@ public class PersistentTopicTest extends MockedBookKeeperTestCase {
verify(nonDeletableSubscription1, times(0)).delete();
verify(deletableSubscription1, times(1)).delete();
verify(nonDeletableSubscription2, times(0)).delete();
+ verify(nonDeletableSubscription3, times(0)).delete();
}
@Test