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 2025/07/17 21:30:45 UTC
(pulsar) 01/02: [fix][broker] Fix wrong backlog age metrics when the mark delete position point to a deleted ledger (#24518)
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 6c8d5587ea0bbc09586bb9706784bd59a624481a
Author: Penghui Li <pe...@apache.org>
AuthorDate: Thu Jul 17 11:42:10 2025 -0700
[fix][broker] Fix wrong backlog age metrics when the mark delete position point to a deleted ledger (#24518)
(cherry picked from commit 7282c06c55c7991123b964d19561e07913bf3ac4)
---
.../broker/service/persistent/PersistentTopic.java | 11 +++++-
.../PersistentTopicProtectedMethodsTest.java | 44 +++++++++++++++++++++-
2 files changed, 53 insertions(+), 2 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 f21a18c56c0..8730b30073a 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
@@ -3630,7 +3630,8 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
Position oldestMarkDeletePosition = oldestMarkDeleteCursorInfo.getPosition();
OldestPositionInfo lastOldestPositionInfo = oldestPositionInfo;
if (lastOldestPositionInfo != null
- && oldestMarkDeletePosition.compareTo(lastOldestPositionInfo.getOldestCursorMarkDeletePosition()) == 0) {
+ && oldestMarkDeletePosition.compareTo(lastOldestPositionInfo.getOldestCursorMarkDeletePosition()) == 0
+ && oldestMarkDeletePosition.compareTo(ledger.getFirstPosition()) >= 0) {
// Same position, but the cursor causing it has changed?
if (!lastOldestPositionInfo.getCursorName().equals(oldestMarkDeleteCursorInfo.getCursor().getName())) {
updateResultIfNewer(new OldestPositionInfo(
@@ -3759,6 +3760,14 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.LedgerInfo
markDeletePositionLedgerInfo = ledger.getLedgerInfo(markDeletePosition.getLedgerId()).get();
+ // If markDeletePositionLedgerInfo is null (ledger no longer exists due to retention/cleanup),
+ // use the next valid position instead to get a meaningful timestamp
+ if (markDeletePositionLedgerInfo == null) {
+ Position nextValidPosition = ledger.getNextValidPosition(markDeletePosition);
+ markDeletePositionLedgerInfo = ledger.getLedgerInfo(nextValidPosition.getLedgerId()).get();
+ markDeletePosition = nextValidPosition;
+ }
+
org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.LedgerInfo positionToCheckLedgerInfo =
markDeletePositionLedgerInfo;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicProtectedMethodsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicProtectedMethodsTest.java
index 1d841483ed7..bba0d2050d8 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicProtectedMethodsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicProtectedMethodsTest.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.broker.service.persistent;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
import java.util.concurrent.CompletableFuture;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.Position;
@@ -51,7 +52,6 @@ public class PersistentTopicProtectedMethodsTest extends ProducerConsumerBase {
}
protected void doInitConf() throws Exception {
- this.conf.setPreciseTimeBasedBacklogQuotaCheck(true);
this.conf.setManagedLedgerMaxEntriesPerLedger(2);
this.conf.setManagedLedgerMaxLedgerRolloverTimeMinutes(10);
this.conf.setManagedLedgerMinLedgerRolloverTimeMinutes(0);
@@ -111,4 +111,46 @@ public class PersistentTopicProtectedMethodsTest extends ProducerConsumerBase {
c1.close();
admin.topics().delete(tp, false);
}
+
+ @Test
+ public void testEstimatedTimeBasedBacklogQuotaCheckWithTopicUnloading() throws Exception {
+ final String tp = BrokerTestUtil.newUniqueName("public/default/tp-with-topic-unloading");
+ admin.topics().createNonPartitionedTopic(tp);
+
+ Consumer<byte[]> c1 = pulsarClient.newConsumer().topic(tp).subscriptionName("s1").subscribe();
+ Producer<byte[]> p1 = pulsarClient.newProducer().topic(tp).create();
+
+ byte[] content = new byte[]{1};
+ for (int i = 0; i < 10; i++) {
+ p1.send(content);
+ }
+
+ PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService().getTopic(tp, false).join().get();
+
+ Awaitility.await().untilAsserted(() -> {
+ admin.brokers().backlogQuotaCheck();
+ assertTrue(persistentTopic.getBestEffortOldestUnacknowledgedMessageAgeSeconds() > 0);
+ });
+
+ for (int i = 0; i < 10; i++) {
+ c1.acknowledge(c1.receive());
+ }
+
+ Awaitility.await().untilAsserted(() -> assertEquals(persistentTopic.getBacklogSize(), 0));
+ admin.topics().unload(tp);
+ for (int i = 0; i < 10; i++) {
+ p1.send(content);
+ }
+
+ PersistentTopic persistentTopicNew = (PersistentTopic) pulsar.getBrokerService()
+ .getTopic(tp, false).join().get();
+ Awaitility.await().untilAsserted(() -> {
+ admin.brokers().backlogQuotaCheck();
+ assertTrue(persistentTopicNew.getBestEffortOldestUnacknowledgedMessageAgeSeconds() > 0);
+ });
+
+ p1.close();
+ c1.close();
+ admin.topics().delete(tp, false);
+ }
}