Skip to content

Commit fe61afd

Browse files
authored
[improve][broker] Skip system cursor when check inactive cursor. (#26149)
1 parent 8576283 commit fe61afd

2 files changed

Lines changed: 8 additions & 1 deletion

File tree

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3700,7 +3700,7 @@ public void checkInactiveSubscriptions(long expirationTimeMillis) {
37003700
subscriptions.forEach((subName, sub) -> {
37013701
if (sub.dispatcher != null && sub.dispatcher.isConsumerConnected()
37023702
|| sub.isReplicated()
3703-
|| isCompactionSubscription(subName)) {
3703+
|| isSystemCursor(subName)) {
37043704
return;
37053705
}
37063706
if (System.currentTimeMillis() - sub.cursor.getLastActive() > expirationTimeMillis) {

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2080,6 +2080,7 @@ public void addFailed(ManagedLedgerException exception, Object ctx) {
20802080
@Test
20812081
public void testCheckInactiveSubscriptions() throws Exception {
20822082
pulsarTestContext.getConfig().setSubscriptionExpirationTimeMinutes(5);
2083+
pulsarTestContext.getConfig().setAdditionalSystemCursorNames(Set.of("additionalSystemCursor"));
20832084
PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService);
20842085

20852086
final var subscriptions = new ConcurrentHashMap<String, PersistentSubscription>();
@@ -2098,6 +2099,11 @@ public void testCheckInactiveSubscriptions() throws Exception {
20982099
spyWithClassAndConstructorArgsRecordingInvocations(PersistentSubscription.class, topic,
20992100
"nonDeletableSubscription2", cursorMock, true);
21002101
subscriptions.put(nonDeletableSubscription2.getName(), nonDeletableSubscription2);
2102+
// This subscription is an additional system cursor.
2103+
PersistentSubscription nonDeletableSubscription3 =
2104+
spyWithClassAndConstructorArgsRecordingInvocations(PersistentSubscription.class, topic,
2105+
"additionalSystemCursor", cursorMock, false);
2106+
subscriptions.put(nonDeletableSubscription3.getName(), nonDeletableSubscription3);
21012107

21022108
Field field = topic.getClass().getDeclaredField("subscriptions");
21032109
field.setAccessible(true);
@@ -2122,6 +2128,7 @@ public void testCheckInactiveSubscriptions() throws Exception {
21222128
verify(nonDeletableSubscription1, times(0)).delete();
21232129
verify(deletableSubscription1, times(1)).delete();
21242130
verify(nonDeletableSubscription2, times(0)).delete();
2131+
verify(nonDeletableSubscription3, times(0)).delete();
21252132
}
21262133

21272134
@Test

0 commit comments

Comments
 (0)