You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zi...@apache.org on 2026/09/09 02:06:04 UTC
(pulsar) branch master updated: [fix][broker] Prevent automatic TTL expiry from skipping reader messages (#26505)
This is an automated email from the ASF dual-hosted git repository.
nodece pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new f3d20b1422fc [fix][broker] Prevent automatic TTL expiry from skipping reader messages (#26505)
f3d20b1422fc is described below
commit f3d20b1422fc2dd1d0b3b168b98ec4929aa969dd
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Sep 9 05:05:57 2026 +0300
[fix][broker] Prevent automatic TTL expiry from skipping reader messages (#26505)
---
.../broker/service/persistent/PersistentTopic.java | 7 +-
.../service/persistent/ReaderMessageTTLTest.java | 99 ++++++++++++++++++++++
2 files changed, 103 insertions(+), 3 deletions(-)
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 e8c3bd7ca89e..274510c8cb93 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
@@ -2410,7 +2410,8 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
private void checkMessageExpiryWithoutSharedPosition(int messageTtlInSeconds) {
subscriptions.forEach((__, sub) -> {
- if (!isCompactionSubscription(sub.getName())
+ // TTL must not advance non-durable reader cursors past unread retained messages.
+ if (sub.getCursor().isDurable() && !isCompactionSubscription(sub.getName())
&& (additionalSystemCursorNames.isEmpty()
|| !additionalSystemCursorNames.contains(sub.getName()))) {
sub.expireMessagesAsync(messageTtlInSeconds);
@@ -2457,9 +2458,9 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
// Nothing need to be expired.
return;
}
- // Expire messages by position, which is more efficient.
+ // Expire durable subscriptions by position. Non-durable readers must retain their read position.
subscriptions.forEach((__, sub) -> {
- if (!isCompactionSubscription(sub.getName())
+ if (sub.getCursor().isDurable() && !isCompactionSubscription(sub.getName())
&& (additionalSystemCursorNames.isEmpty()
|| !additionalSystemCursorNames.contains(sub.getName()))) {
// The variable "position" is to mark delete position.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReaderMessageTTLTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReaderMessageTTLTest.java
new file mode 100644
index 000000000000..2e57a883aca0
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReaderMessageTTLTest.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.persistent;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.AdditionalAnswers.delegatesTo;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import java.util.concurrent.TimeUnit;
+import org.apache.bookkeeper.mledger.ManagedLedger;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.pulsar.broker.service.SharedPulsarBaseTest;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.common.policies.data.RetentionPolicies;
+import org.awaitility.Awaitility;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class ReaderMessageTTLTest extends SharedPulsarBaseTest {
+
+ @DataProvider
+ public Object[][] expiryPaths() {
+ return new Object[][] {{true}, {false}};
+ }
+
+ @Test(dataProvider = "expiryPaths", timeOut = 60000)
+ public void testReaderRetainsUnreadMessagesDuringExpiry(boolean sharedPosition) throws Exception {
+ String topicName = newTopicName();
+ admin.namespaces().setRetention(getNamespace(), new RetentionPolicies(-1, -1));
+ admin.namespaces().setNamespaceMessageTTL(getNamespace(), 1);
+ admin.topics().createNonPartitionedTopic(topicName);
+ admin.topics().createSubscription(topicName, "durable", MessageId.earliest);
+ int messageCount = 20;
+ try (Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName)
+ .enableBatching(false).create()) {
+ for (int i = 0; i < messageCount; i++) {
+ producer.send("message-" + i);
+ }
+ }
+ try (Reader<String> reader = pulsarClient.newReader(Schema.STRING).topic(topicName)
+ .subscriptionName("reader").startMessageId(MessageId.earliest).receiverQueueSize(1).create()) {
+ Message<String> first = reader.readNext(10, TimeUnit.SECONDS);
+ assertThat(first).isNotNull();
+ assertThat(first.getValue()).isEqualTo("message-0");
+ PersistentTopic topic = (PersistentTopic) getTopicReference(topicName).orElseThrow();
+ PersistentSubscription readerSubscription = topic.getSubscription("reader");
+ assertThat(readerSubscription.getCursor().isDurable()).isFalse();
+ Position readPosition = readerSubscription.getCursor().getReadPosition();
+ PersistentSubscription durable = topic.getSubscription("durable");
+
+ // Pause the reader until all published messages are old enough to expire.
+ long expiryTime = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(2);
+ Awaitility.await().until(() -> System.currentTimeMillis() > expiryTime);
+ PersistentTopic expiryTopic = topic;
+ if (!sharedPosition) {
+ expiryTopic = spy(topic);
+ // Exercise the fallback used by custom ManagedLedger implementations.
+ doReturn(mock(ManagedLedger.class, delegatesTo(topic.getManagedLedger())))
+ .when(expiryTopic).getManagedLedger();
+ }
+ PersistentTopic topicToCheck = expiryTopic;
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ topicToCheck.checkMessageExpiry();
+ assertThat(durable.getCursor().getNumberOfEntriesInBacklog(false)).isZero();
+ });
+ assertThat(readerSubscription.getCursor().getMarkDeletedPosition())
+ .as("TTL must not acknowledge unread reader messages")
+ .isLessThan(readPosition);
+ for (int i = 1; i < messageCount; i++) {
+ Message<String> message = reader.readNext(10, TimeUnit.SECONDS);
+ assertThat(message).as("retained message %s", i).isNotNull();
+ assertThat(message.getValue()).isEqualTo("message-" + i);
+ }
+ assertThat(reader.hasMessageAvailable()).isFalse();
+ }
+ }
+}