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/04/28 12:00:02 UTC
(pulsar) 03/04: [improve][broker]Improve the feature "Optimize subscription seek (cursor reset) by timestamp": search less entries (#24219)
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 376bf6e94477cb07fcadfbdd26a5ca445ce27192
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Sun Apr 27 16:13:01 2025 +0800
[improve][broker]Improve the feature "Optimize subscription seek (cursor reset) by timestamp": search less entries (#24219)
(cherry picked from commit bb2e4ab897e3827f8869cef349bdfbe092531fec)
---
.../service/persistent/PersistentMessageFinder.java | 15 ++++++++++++++-
.../broker/service/PersistentMessageFinderTest.java | 19 ++++++++++---------
2 files changed, 24 insertions(+), 10 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageFinder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageFinder.java
index e780f167284..74a9d713e77 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageFinder.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageFinder.java
@@ -72,6 +72,7 @@ public class PersistentMessageFinder implements AsyncCallbacks.FindEntryCallback
ledgerCloseTimestampMaxClockSkewMillis);
cursor.asyncFindNewestMatching(ManagedCursor.FindPositionConstraint.SearchAllAvailableEntries, entry -> {
try {
+ // Find the latest entry that is earlier than the target timestamp.
long entryTimestamp = Commands.getEntryTimestamp(entry.getDataBuffer());
return MessageImpl.isEntryPublishedEarlierThan(entryTimestamp, timestamp);
} catch (Exception e) {
@@ -92,6 +93,12 @@ public class PersistentMessageFinder implements AsyncCallbacks.FindEntryCallback
}
}
+ /**
+ * The range may be across multi ledgers:
+ * - start: the latest ledger that closed before {@param targetTimestamp}.
+ * - only the latest entry is useful.
+ * - end: the earliest ledger that is larger than the target timestamp.
+ */
@VisibleForTesting
public static Pair<Position, Position> getFindPositionRange(Iterable<LedgerInfo> ledgerInfos,
Position lastConfirmedEntry, long targetTimestamp,
@@ -107,6 +114,10 @@ public class PersistentMessageFinder implements AsyncCallbacks.FindEntryCallback
Position start = null;
Position end = null;
+ // We do not use binary search here:
+ // Since "managedLedger.ledgers" os a map, we can hardly use a binary search except to copy items to an array,
+ // which causes frequently young GC. And "collection.toArray()" also loops the collection once, which does not
+ // benefit performance anymore.
for (LedgerInfo info : ledgerInfos) {
if (!info.hasTimestamp()) {
// unexpected case, don't set start and end
@@ -119,7 +130,9 @@ public class PersistentMessageFinder implements AsyncCallbacks.FindEntryCallback
break;
}
if (closeTimestamp <= targetTimestampMin) {
- start = PositionFactory.create(info.getLedgerId(), 0);
+ // Since we have "broker.conf -> managedLedgerCursorResetLedgerCloseTimestampMaxClockSkewMillis", which
+ // already expanded the scope for searching, the entries before the latest one is not useful.
+ start = PositionFactory.create(info.getLedgerId(), info.getEntries() - 1);
} else if (closeTimestamp > targetTimestampMax) {
// If the close timestamp is greater than the timestamp
end = PositionFactory.create(info.getLedgerId(), info.getEntries() - 1);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java
index 62e27eaea41..bab21fbc001 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java
@@ -618,7 +618,7 @@ public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
assertNotNull(range);
assertNotNull(range.getLeft());
assertNull(range.getRight());
- assertEquals(range.getLeft(), PositionFactory.create(2, 0));
+ assertEquals(range.getLeft(), PositionFactory.create(2, 9));
}
@Test
@@ -636,7 +636,7 @@ public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
assertNotNull(range);
assertNotNull(range.getLeft());
assertNull(range.getRight());
- assertEquals(range.getLeft(), PositionFactory.create(2, 0));
+ assertEquals(range.getLeft(), PositionFactory.create(2, 9));
}
@Test
@@ -654,7 +654,8 @@ public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
assertNotNull(range);
assertNotNull(range.getLeft());
assertNull(range.getRight());
- assertEquals(range.getLeft(), PositionFactory.create(2, 0));
+ assertEquals(range.getLeft(), PositionFactory.create(2, 9));
+ assertEquals(range.getLeft(), PositionFactory.create(2, 9));
}
@Test
@@ -689,7 +690,7 @@ public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
assertNotNull(range);
assertNotNull(range.getLeft());
assertNotNull(range.getRight());
- assertEquals(range.getLeft(), PositionFactory.create(2, 0));
+ assertEquals(range.getLeft(), PositionFactory.create(2, 9));
assertEquals(range.getRight(), PositionFactory.create(3, 9));
}
@@ -709,7 +710,7 @@ public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
assertNotNull(range);
assertNotNull(range.getLeft());
assertNotNull(range.getRight());
- assertEquals(range.getLeft(), PositionFactory.create(3, 0));
+ assertEquals(range.getLeft(), PositionFactory.create(3, 9));
// there might be entries in the next ledger with the same timestamp as the target timestamp, even though
// the close timestamp of ledger 3 is equals to the target timestamp
assertEquals(range.getRight(), PositionFactory.create(4, 9));
@@ -732,7 +733,7 @@ public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
assertNotNull(range);
assertNotNull(range.getLeft());
assertNotNull(range.getRight());
- assertEquals(range.getLeft(), PositionFactory.create(1, 0));
+ assertEquals(range.getLeft(), PositionFactory.create(1, 9));
assertEquals(range.getRight(), PositionFactory.create(4, 9));
}
@@ -753,7 +754,7 @@ public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
assertNotNull(range);
assertNotNull(range.getLeft());
assertNotNull(range.getRight());
- assertEquals(range.getLeft(), PositionFactory.create(2, 0));
+ assertEquals(range.getLeft(), PositionFactory.create(2, 9));
assertEquals(range.getRight(), PositionFactory.create(4, 9));
}
@@ -774,7 +775,7 @@ public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
assertNotNull(range);
assertNotNull(range.getLeft());
assertNotNull(range.getRight());
- assertEquals(range.getLeft(), PositionFactory.create(2, 0));
+ assertEquals(range.getLeft(), PositionFactory.create(2, 9));
assertEquals(range.getRight(), PositionFactory.create(4, 9));
}
@@ -825,6 +826,6 @@ public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
assertNotNull(range);
assertNotNull(range.getLeft());
assertNull(range.getRight());
- assertEquals(range.getLeft(), PositionFactory.create(1, 0));
+ assertEquals(range.getLeft(), PositionFactory.create(1, 9));
}
}