[fix] [broker] Expire messages according to ledger close time to avoid client clock skew

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

@@ -20,6 +20,7 @@
 
 import java.util.Objects;
 import java.util.Optional;
+import java.util.SortedMap;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.atomic.LongAdder;
 import javax.annotation.Nullable;
@@ -30,8 +31,10 @@
 import org.apache.bookkeeper.mledger.ManagedLedgerException.LedgerNotExistException;
 import org.apache.bookkeeper.mledger.ManagedLedgerException.NonRecoverableLedgerException;
 import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
 import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
 import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.mledger.proto.MLDataFormats;
 import org.apache.pulsar.broker.service.MessageExpirer;
 import org.apache.pulsar.client.impl.MessageImpl;
 import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
@@ -78,7 +81,9 @@ public boolean expireMessages(int messageTTLInSeconds) {
         if (expirationCheckInProgressUpdater.compareAndSet(this, FALSE, TRUE)) {
             log.info("[{}][{}] Starting message expiry check, ttl= {} seconds", topicName, subName,
                     messageTTLInSeconds);
-
+            if (checkExpiryByLedgerClosureTime(cursor, messageTTLInSeconds)) {
+                return true;
+            }
             cursor.asyncFindNewestMatching(ManagedCursor.FindPositionConstraint.SearchActiveEntries, entry -> {
                 try {
                     long entryTimestamp = Commands.getEntryTimestamp(entry.getDataBuffer());
@@ -100,6 +105,27 @@ public boolean expireMessages(int messageTTLInSeconds) {
         }
     }
 
+    private boolean checkExpiryByLedgerClosureTime(ManagedCursor cursor, int messageTTLInSeconds) {
+        if (cursor instanceof ManagedCursorImpl managedCursor) {
+            ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) managedCursor.getManagedLedger();
+            Position deletedPosition = managedCursor.getMarkDeletedPosition();
+            SortedMap<Long, MLDataFormats.ManagedLedgerInfo.LedgerInfo> ledgerInfoSortedMap =
+                    managedLedger.getLedgersInfo()
+                            .subMap(deletedPosition.getLedgerId(), managedLedger.getLedgersInfo().lastKey());
+            long expiryLedgerId = -1;
+            for (MLDataFormats.ManagedLedgerInfo.LedgerInfo ledgerInfo : ledgerInfoSortedMap.values()) {
+                if (!MessageImpl.isEntryExpired(messageTTLInSeconds, ledgerInfo.getTimestamp())) {
+                    break;
+                }
+                expiryLedgerId = ledgerInfo.getLedgerId();
+            }
+            if (expiryLedgerId > -1) {
+                findEntryComplete(PositionImpl.get(managedLedger.getNextValidLedger(expiryLedgerId), 0), null);
+            }
+        }
+        return false;
+    }
+
     @Override
     public boolean expireMessages(Position messagePosition) {
         // If it's beyond last position of this topic, do nothing.

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

@@ -81,8 +81,11 @@
 public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
 
     public static byte[] createMessageWrittenToLedger(String msg) {
+        return createMessageWrittenToLedger(msg, System.currentTimeMillis());
+    }
+    public static byte[] createMessageWrittenToLedger(String msg, long messageTimestamp) {
         MessageMetadata messageMetadata = new MessageMetadata()
-                    .setPublishTime(System.currentTimeMillis())
+                    .setPublishTime(messageTimestamp)
                     .setProducerName("createMessageWrittenToLedger")
                     .setSequenceId(1);
         ByteBuf data = UnpooledByteBufAllocator.DEFAULT.heapBuffer().writeBytes(msg.getBytes());
@@ -437,6 +440,28 @@ void testMessageExpiryWithTimestampNonRecoverableException() throws Exception {
 
     }
 
+    @Test
+    public void testIncorrectClientClock() throws Exception {
+        final String ledgerAndCursorName = "testIncorrectClientClock";
+        int maxRollOverTimeMs = 1;
+        int maxTTLSeconds = 1;
+        ManagedLedgerConfig config = new ManagedLedgerConfig();
+        config.setMaximumRolloverTime(maxRollOverTimeMs, TimeUnit.MICROSECONDS);
+        ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(ledgerAndCursorName, config);
+        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor(ledgerAndCursorName);
+        // set client clock to 10 days later
+        long incorrectPublishTimestamp = System.currentTimeMillis() + TimeUnit.DAYS.toMillis(10);
+        for (int i = 0; i < 10; i++) {
+            ledger.addEntry(createMessageWrittenToLedger("msg" + i, incorrectPublishTimestamp));
+        }
+        PersistentTopic mock = mock(PersistentTopic.class);
+        when(mock.getName()).thenReturn("topicname");
+        when(mock.getLastPosition()).thenReturn(PositionImpl.EARLIEST);
+        PersistentMessageExpiryMonitor monitor = new PersistentMessageExpiryMonitor(mock, c1.getName(), c1, null);
+        Thread.sleep(Math.max(config.getMaximumRolloverTimeMs(), TimeUnit.SECONDS.toMillis(maxTTLSeconds)));
+        assertTrue(monitor.expireMessages(maxTTLSeconds));
+    }
+
     @Test
     void testMessageExpiryWithPosition() throws Exception {
         final String ledgerAndCursorName = "testPersistentMessageExpiryWithPositionNonRecoverableLedgers";