[improve][broker] optimize the problem of subscription snapshot cache not hitting

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java

@@ -211,7 +211,7 @@ public int filterEntriesForConsumer(@Nullable MessageMetadata[] metadataArray, i
 
                 if (Markers.isReplicatedSubscriptionSnapshotMarker(msgMetadata)) {
                     final int readerIndex = metadataAndPayload.readerIndex();
-                    processReplicatedSubscriptionSnapshot(pos, metadataAndPayload);
+                    processReplicatedSubscriptionSnapshot(pos, metadataAndPayload, msgMetadata.getPublishTime());
                     metadataAndPayload.readerIndex(readerIndex);
                 }
 
@@ -358,13 +358,13 @@ protected boolean isConsumersExceededOnSubscription(AbstractTopic topic, int con
                 && maxConsumersPerSubscription <= consumerSize;
     }
 
-    private void processReplicatedSubscriptionSnapshot(Position pos, ByteBuf headersAndPayload) {
+    private void processReplicatedSubscriptionSnapshot(Position pos, ByteBuf headersAndPayload, long publishTime) {
         // Remove the protobuf headers
         Commands.skipMessageMetadata(headersAndPayload);
 
         try {
             ReplicatedSubscriptionsSnapshot snapshot = Markers.parseReplicatedSubscriptionsSnapshot(headersAndPayload);
-            subscription.processReplicatedSubscriptionSnapshot(snapshot);
+            subscription.processReplicatedSubscriptionSnapshot(snapshot, publishTime);
         } catch (Throwable t) {
             log.warn("Failed to process replicated subscription snapshot at {} -- {}", pos, t.getMessage(), t);
             return;

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Subscription.java

@@ -106,7 +106,7 @@ default long getNumberOfEntriesDelayed() {
 
     boolean isSubscriptionMigrated();
 
-    default void processReplicatedSubscriptionSnapshot(ReplicatedSubscriptionsSnapshot snapshot) {
+    default void processReplicatedSubscriptionSnapshot(ReplicatedSubscriptionsSnapshot snapshot, long publishTime) {
         // Default is no-op
     }
 

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

@@ -210,7 +210,8 @@ public boolean setReplicated(boolean replicated) {
             this.replicatedSubscriptionSnapshotCache = null;
         } else if (this.replicatedSubscriptionSnapshotCache == null) {
             this.replicatedSubscriptionSnapshotCache = new ReplicatedSubscriptionSnapshotCache(subName,
-                    config.getReplicatedSubscriptionsSnapshotMaxCachedPerSubscription());
+                    config.getReplicatedSubscriptionsSnapshotMaxCachedPerSubscription(),
+                    config.getReplicatedSubscriptionsSnapshotFrequencyMillis());
         }
 
         if (this.cursor != null) {
@@ -1523,10 +1524,10 @@ protected Map<String, Long> mergeCursorProperties(Map<String, Long> userProperti
     }
 
     @Override
-    public void processReplicatedSubscriptionSnapshot(ReplicatedSubscriptionsSnapshot snapshot) {
+    public void processReplicatedSubscriptionSnapshot(ReplicatedSubscriptionsSnapshot snapshot, long publishTime) {
         ReplicatedSubscriptionSnapshotCache snapshotCache = this.replicatedSubscriptionSnapshotCache;
         if (snapshotCache != null) {
-            snapshotCache.addNewSnapshot(new ReplicatedSubscriptionsSnapshot().copyFrom(snapshot));
+            snapshotCache.addNewSnapshot(new ReplicatedSubscriptionsSnapshot().copyFrom(snapshot), publishTime);
         }
     }
 

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

@@ -18,6 +18,8 @@
  */
 package org.apache.pulsar.broker.service.persistent;
 
+import java.util.Iterator;
+import java.util.Map;
 import java.util.NavigableMap;
 import java.util.TreeMap;
 import lombok.extern.slf4j.Slf4j;
@@ -34,28 +36,110 @@ public class ReplicatedSubscriptionSnapshotCache {
     private final String subscription;
     private final NavigableMap<Position, ReplicatedSubscriptionsSnapshot> snapshots;
     private final int maxSnapshotToCache;
+    private final int snapshotFrequencyMillis;
 
-    public ReplicatedSubscriptionSnapshotCache(String subscription, int maxSnapshotToCache) {
+    public ReplicatedSubscriptionSnapshotCache(String subscription, int maxSnapshotToCache,
+                                               int snapshotFrequencyMillis) {
         this.subscription = subscription;
         this.snapshots = new TreeMap<>();
         this.maxSnapshotToCache = maxSnapshotToCache;
+        this.snapshotFrequencyMillis = snapshotFrequencyMillis;
     }
 
-    public synchronized void addNewSnapshot(ReplicatedSubscriptionsSnapshot snapshot) {
+    /**
+     * Adds a new replicated subscription snapshot to the cache with time-based eviction policies.
+     * This method handles 5 distinct cases for cache management:
+     * <ol>
+     *   <li><b>Empty Cache</b>: Directly inserts the snapshot if cache is empty.</li>
+     *   <li><b>Position Precedes First Entry</b>: Clears entire cache and inserts new snapshot.</li>
+     *   <li><b>Position in Middle</b>: Trims later entries and inserts new snapshot.</li>
+     *   <li><b>Cache Not Full</b>: Inserts snapshot if cache has available space.</li>
+     *   <li><b>Cache Full</b>: Evicts median-aged entry before insertion (time-window optimized).</li>
+     * </ol>
+     *
+     * <p>Time-based eviction considers two conditions:
+     * <ul>
+     *   <li>Minimum frequency interval ({@code snapshotFrequencyMillis})</li>
+     *   <li>Dynamic time window per slot ({@code timeWindowPerSlot})</li>
+     * </ul>
+     *
+     * @param snapshot    The replicated subscription snapshot to add (non-null)
+     * @param publishTime The ReplicatedSubscriptionsSnapshot marker message (entry),
+     *                    the publishTime originates from the same broker, the time is reliable and clock skew is
+     *                    not a problem.
+     * @throws NullPointerException If snapshot argument is null
+     * @see ReplicatedSubscriptionsSnapshot
+     * @see #findPositionByIndex(int)  For median position calculation in eviction case
+     */
+    public synchronized void addNewSnapshot(ReplicatedSubscriptionsSnapshot snapshot, long publishTime) {
+        snapshot.setMarkersTimestamp(publishTime);
         MarkersMessageIdData msgId = snapshot.getLocalMessageId();
         Position position = PositionFactory.create(msgId.getLedgerId(), msgId.getEntryId());
 
         if (log.isDebugEnabled()) {
             log.debug("[{}] Added new replicated-subscription snapshot at {} -- {}", subscription, position,
                     snapshot.getSnapshotId());
         }
+        // Case 1: cache if empty
+        if (snapshots.lastEntry() == null) {
+            snapshots.put(position, snapshot);
+            return;
+        }
 
-        snapshots.put(position, snapshot);
+        // The time difference between the previous position and the earliest cache entry
+        final long timeSinceFirstSnapshot = publishTime - snapshots.firstEntry().getValue().getMarkersTimestamp();
+        // The time difference between the previous position and the lately cache entry
+        final long timeSinceLastSnapshot = publishTime - snapshots.lastEntry().getValue().getMarkersTimestamp();
+        // The time window length of each time slot, used for dynamic adjustment in the snapshot cache.
+        // The larger the time slot, the slower the update.
+        final long timeWindowPerSlot = timeSinceFirstSnapshot / snapshotFrequencyMillis / maxSnapshotToCache;
 
-        // Prune the cache
-        while (snapshots.size() > maxSnapshotToCache) {
-            snapshots.pollFirstEntry();
+        if (position.compareTo(snapshots.firstKey()) < 0) {
+            // Case 2: Reset cursor if position precedes first entry
+            snapshots.clear();
+            snapshots.put(position, snapshot);
+            return;
+        } else if (position.compareTo(snapshots.lastKey()) < 0) {
+            // Case 3: Reset cursor If the position is in the middle, delete the cache after that position
+            while (position.compareTo(snapshots.lastKey()) < 0) {
+                snapshots.pollLastEntry();
+            }
+            snapshots.put(position, snapshot);
+        }
+        // Time-based eviction conditions
+        // timeSinceLastSnapshot < snapshotFrequencyMillis, keep the same frequency
+        // timeSinceLastSnapshot < timeWindowPerSlot, implementing dynamic adjustments
+        if (timeSinceLastSnapshot < snapshotFrequencyMillis || timeSinceLastSnapshot < timeWindowPerSlot) {
+            return;
+        }
+        if (snapshots.size() < maxSnapshotToCache) {
+            // Case 4: Add to cache if not full
+            snapshots.put(position, snapshot);
+        } else {
+            // Case 5: Median-based eviction when cache is full
+            int medianIndex = maxSnapshotToCache / 2;
+            Position positionToRemove = findPositionByIndex(medianIndex);
+            if (positionToRemove != null) {
+                snapshots.remove(positionToRemove);
+            }
+            snapshots.put(position, snapshot);
+        }
+    }
+
+    /**
+     * Find the Position in NavigableMap according to the target index.
+     */
+    private Position findPositionByIndex(int targetIndex) {
+        Iterator<Map.Entry<Position, ReplicatedSubscriptionsSnapshot>> it = snapshots.entrySet().iterator();
+        int currentIndex = 0;
+        while (it.hasNext()) {
+            Map.Entry<Position, ReplicatedSubscriptionsSnapshot> entry = it.next();
+            if (currentIndex == targetIndex) {
+                return entry.getKey();
+            }
+            currentIndex++;
         }
+        return null;
     }
 
     /**

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionSnapshotCacheTest.java

@@ -28,10 +28,13 @@
 
 @Test(groups = "broker")
 public class ReplicatedSubscriptionSnapshotCacheTest {
+    int snapshotFrequencyMillis = 1000;
 
     @Test
     public void testSnapshotCache() {
-        ReplicatedSubscriptionSnapshotCache cache = new ReplicatedSubscriptionSnapshotCache("my-subscription", 10);
+
+        ReplicatedSubscriptionSnapshotCache cache =
+                new ReplicatedSubscriptionSnapshotCache("my-subscription", 10, snapshotFrequencyMillis);
 
         assertNull(cache.advancedMarkDeletePosition(PositionFactory.create(0, 0)));
         assertNull(cache.advancedMarkDeletePosition(PositionFactory.create(100, 0)));
@@ -51,11 +54,11 @@ public void testSnapshotCache() {
         ReplicatedSubscriptionsSnapshot s7 = new ReplicatedSubscriptionsSnapshot()
                 .setSnapshotId("snapshot-7");
         s7.setLocalMessageId().setLedgerId(7 ).setEntryId(7);
-
-        cache.addNewSnapshot(s1);
-        cache.addNewSnapshot(s2);
-        cache.addNewSnapshot(s5);
-        cache.addNewSnapshot(s7);
+        long publishTime = System.currentTimeMillis();
+        cache.addNewSnapshot(s1, publishTime);
+        cache.addNewSnapshot(s2, publishTime + snapshotFrequencyMillis);
+        cache.addNewSnapshot(s5, publishTime + 2L * snapshotFrequencyMillis);
+        cache.addNewSnapshot(s7, publishTime + 3L * snapshotFrequencyMillis);
 
         assertNull(cache.advancedMarkDeletePosition(PositionFactory.create(0, 0)));
         assertNull(cache.advancedMarkDeletePosition(PositionFactory.create(1, 0)));
@@ -72,9 +75,73 @@ public void testSnapshotCache() {
         assertNull(cache.advancedMarkDeletePosition(PositionFactory.create(5, 5)));
     }
 
+    @Test
+    public void testSnapshotCacheByRestCursor() {
+
+        ReplicatedSubscriptionSnapshotCache cache =
+                new ReplicatedSubscriptionSnapshotCache("my-subscription", 10, snapshotFrequencyMillis);
+        // The rest cursor is smaller than the cache first position
+        ReplicatedSubscriptionsSnapshot s1 = new ReplicatedSubscriptionsSnapshot()
+                .setSnapshotId("snapshot-1");
+        s1.setLocalMessageId().setLedgerId(10).setEntryId(10);
+
+        ReplicatedSubscriptionsSnapshot s2 = new ReplicatedSubscriptionsSnapshot()
+                .setSnapshotId("snapshot-2");
+        s2.setLocalMessageId().setLedgerId(20).setEntryId(20);
+
+        ReplicatedSubscriptionsSnapshot s3 = new ReplicatedSubscriptionsSnapshot()
+                .setSnapshotId("snapshot-3");
+        s3.setLocalMessageId().setLedgerId(30).setEntryId(30);
+
+        ReplicatedSubscriptionsSnapshot s4 = new ReplicatedSubscriptionsSnapshot()
+                .setSnapshotId("snapshot-4");
+        s4.setLocalMessageId().setLedgerId(1).setEntryId(1);
+
+        ReplicatedSubscriptionsSnapshot s5 = new ReplicatedSubscriptionsSnapshot()
+                .setSnapshotId("snapshot-5");
+        s5.setLocalMessageId().setLedgerId(10).setEntryId(10);
+        long publishTime = System.currentTimeMillis();
+        cache.addNewSnapshot(s1, publishTime);
+        cache.addNewSnapshot(s2, publishTime + 2L * snapshotFrequencyMillis);
+        cache.addNewSnapshot(s3, publishTime + 3L * snapshotFrequencyMillis);
+        cache.addNewSnapshot(s4, publishTime + 4L * snapshotFrequencyMillis);
+        cache.addNewSnapshot(s5, publishTime + 5L * snapshotFrequencyMillis);
+
+        ReplicatedSubscriptionsSnapshot snapshot = cache.advancedMarkDeletePosition(PositionFactory.create(1, 1));
+        assertEquals(snapshot.getSnapshotId(), "snapshot-4");
+        snapshot = cache.advancedMarkDeletePosition(PositionFactory.create(10, 10));
+        assertEquals(snapshot.getSnapshotId(), "snapshot-5");
+        assertNull(cache.advancedMarkDeletePosition(PositionFactory.create(20, 20)));
+        assertNull(cache.advancedMarkDeletePosition(PositionFactory.create(30, 30)));
+
+        // The rest cursor is smaller than the cache last position
+        ReplicatedSubscriptionsSnapshot s6 = new ReplicatedSubscriptionsSnapshot()
+                .setSnapshotId("snapshot-6");
+        s6.setLocalMessageId().setLedgerId(10).setEntryId(10);
+        ReplicatedSubscriptionsSnapshot s7 = new ReplicatedSubscriptionsSnapshot()
+                .setSnapshotId("snapshot-7");
+        s7.setLocalMessageId().setLedgerId(20).setEntryId(20);
+        ReplicatedSubscriptionsSnapshot s8 = new ReplicatedSubscriptionsSnapshot()
+                .setSnapshotId("snapshot-8");
+        s8.setLocalMessageId().setLedgerId(30).setEntryId(30);
+        ReplicatedSubscriptionsSnapshot s9 = new ReplicatedSubscriptionsSnapshot()
+                .setSnapshotId("snapshot-9");
+        s9.setLocalMessageId().setLedgerId(20).setEntryId(20);
+        cache.addNewSnapshot(s6, publishTime + 6L * snapshotFrequencyMillis);
+        cache.addNewSnapshot(s7, publishTime + 7L * snapshotFrequencyMillis);
+        cache.addNewSnapshot(s8, publishTime + 8L * snapshotFrequencyMillis);
+        cache.addNewSnapshot(s9, publishTime + 9L * snapshotFrequencyMillis);
+        snapshot = cache.advancedMarkDeletePosition(PositionFactory.create(10, 10));
+        assertEquals(snapshot.getSnapshotId(), "snapshot-6");
+        snapshot = cache.advancedMarkDeletePosition(PositionFactory.create(20, 20));
+        assertEquals(snapshot.getSnapshotId(), "snapshot-9");
+        assertNull(cache.advancedMarkDeletePosition(PositionFactory.create(30, 30)));
+    }
+
     @Test
     public void testSnapshotCachePruning() {
-        ReplicatedSubscriptionSnapshotCache cache = new ReplicatedSubscriptionSnapshotCache("my-subscription", 3);
+        ReplicatedSubscriptionSnapshotCache cache =
+                new ReplicatedSubscriptionSnapshotCache("my-subscription", 4, snapshotFrequencyMillis);
 
         ReplicatedSubscriptionsSnapshot s1 = new ReplicatedSubscriptionsSnapshot()
                 .setSnapshotId("snapshot-1");
@@ -91,20 +158,37 @@ public void testSnapshotCachePruning() {
         ReplicatedSubscriptionsSnapshot s4 = new ReplicatedSubscriptionsSnapshot()
                 .setSnapshotId("snapshot-4");
         s4.setLocalMessageId().setLedgerId(4).setEntryId(4);
-
-        cache.addNewSnapshot(s1);
-        cache.addNewSnapshot(s2);
-        cache.addNewSnapshot(s3);
-        cache.addNewSnapshot(s4);
-
-        // Snapshot-1 was already pruned
-        assertNull(cache.advancedMarkDeletePosition(PositionFactory.create(1, 1)));
+        ReplicatedSubscriptionsSnapshot s5 = new ReplicatedSubscriptionsSnapshot()
+                .setSnapshotId("snapshot-5");
+        s5.setLocalMessageId().setLedgerId(5).setEntryId(5);
+        ReplicatedSubscriptionsSnapshot s6 = new ReplicatedSubscriptionsSnapshot()
+                .setSnapshotId("snapshot-6");
+        s6.setLocalMessageId().setLedgerId(6).setEntryId(6);
+        ReplicatedSubscriptionsSnapshot s7 = new ReplicatedSubscriptionsSnapshot()
+                .setSnapshotId("snapshot-7");
+        s7.setLocalMessageId().setLedgerId(7).setEntryId(7);
+
+        long publishTime = System.currentTimeMillis();
+        cache.addNewSnapshot(s1, publishTime + snapshotFrequencyMillis);
+        cache.addNewSnapshot(s2, publishTime + 2L * snapshotFrequencyMillis);
+        cache.addNewSnapshot(s3, publishTime + 3L * snapshotFrequencyMillis);
+        cache.addNewSnapshot(s4, publishTime + 4L * snapshotFrequencyMillis);
+        cache.addNewSnapshot(s5, publishTime + 5L * snapshotFrequencyMillis);
+        cache.addNewSnapshot(s6, publishTime + 5L * snapshotFrequencyMillis);
+        cache.addNewSnapshot(s7, publishTime + 7L * snapshotFrequencyMillis);
+        // snapshots = [s1, s2, s5, s7]
+        cache.advancedMarkDeletePosition(PositionFactory.create(1, 1));
         ReplicatedSubscriptionsSnapshot snapshot = cache.advancedMarkDeletePosition(PositionFactory.create(2, 2));
-        assertNotNull(snapshot);
         assertEquals(snapshot.getSnapshotId(), "snapshot-2");
-
+        snapshot = cache.advancedMarkDeletePosition(PositionFactory.create(3, 3));
+        assertNull(snapshot);
+        snapshot = cache.advancedMarkDeletePosition(PositionFactory.create(4, 4));
+        assertNull(snapshot);
         snapshot = cache.advancedMarkDeletePosition(PositionFactory.create(5, 5));
-        assertNotNull(snapshot);
-        assertEquals(snapshot.getSnapshotId(), "snapshot-4");
+        assertEquals(snapshot.getSnapshotId(), "snapshot-5");
+        snapshot = cache.advancedMarkDeletePosition(PositionFactory.create(6, 6));
+        assertNull(snapshot);
+        snapshot = cache.advancedMarkDeletePosition(PositionFactory.create(7, 7));
+        assertEquals(snapshot.getSnapshotId(), "snapshot-7");
     }
 }

pulsar-common/src/main/proto/PulsarMarkers.proto

@@ -60,6 +60,7 @@ message ReplicatedSubscriptionsSnapshot {
     required string snapshot_id = 1;
     optional MarkersMessageIdData local_message_id = 2;
     repeated ClusterMessageId clusters = 3;
+    optional int64 markers_timestamp = 4;
 }
 
 // When the replicated subscription mark-delete position