[improve][broker]Improve PersistentMessageExpiryMonitor expire speed when ledger not existed

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerException.java

@@ -142,9 +142,19 @@ public TooManyRequestsException(String msg) {
     }
 
     public static class NonRecoverableLedgerException extends ManagedLedgerException {
+        private Integer bkErrorCode; // null means ledger not exists or deleted
         public NonRecoverableLedgerException(String msg) {
             super(msg);
         }
+
+        public NonRecoverableLedgerException(String msg, Integer bkErrorCode) {
+            super(msg);
+            this.bkErrorCode = bkErrorCode;
+        }
+
+        public Integer getBkErrorCode() {
+            return bkErrorCode;
+        }
     }
 
     public static class InvalidReplayPositionException extends ManagedLedgerException {

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java

@@ -3760,7 +3760,7 @@ public static ManagedLedgerException createManagedLedgerException(int bkErrorCod
         if (bkErrorCode == BKException.Code.TooManyRequestsException) {
             return new TooManyRequestsException("Too many request error from bookies");
         } else if (isBkErrorNotRecoverable(bkErrorCode)) {
-            return new NonRecoverableLedgerException(BKException.getMessage(bkErrorCode));
+            return new NonRecoverableLedgerException(BKException.getMessage(bkErrorCode), bkErrorCode);
         } else {
             return new ManagedLedgerException(BKException.getMessage(bkErrorCode));
         }

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

@@ -22,6 +22,7 @@
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.atomic.LongAdder;
+import org.apache.bookkeeper.client.BKException;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.FindEntryCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.MarkDeleteCallback;
 import org.apache.bookkeeper.mledger.ManagedCursor;
@@ -191,9 +192,38 @@ public void findEntryFailed(ManagedLedgerException exception, Optional<Position>
                 && (exception instanceof NonRecoverableLedgerException)) {
             log.warn("[{}][{}] read failed from ledger at position:{} : {}", topicName, subName, failedReadPosition,
                     exception.getMessage());
-            findEntryComplete(failedReadPosition.get(), ctx);
+            if (isLedgerNotExistException(((NonRecoverableLedgerException) exception).getBkErrorCode())) {
+                try {
+                    long failedLedgerId = failedReadPosition.get().getLedgerId();
+                    Position lastPositionInLedger = PositionImpl.get(failedLedgerId,
+                            cursor.getManagedLedger().getLedgerInfo(failedLedgerId).get().getEntries() - 1);
+                    log.info("[{}][{}] ledger not existed, will complete the last position of the non-existed"
+                                    + " ledger:{}", topicName, subName, lastPositionInLedger);
+                    findEntryComplete(lastPositionInLedger, ctx);
+                } catch (Exception e) {
+                  log.warn("[{}][{}] failed to expire not existed ledger, try to just expire failed position:{} : {}",
+                          topicName, subName, failedReadPosition, e.getMessage());
+                    findEntryComplete(failedReadPosition.get(), ctx);
+                }
+            } else {
+                findEntryComplete(failedReadPosition.get(), ctx);
+            }
         }
         expirationCheckInProgress = FALSE;
         updateRates();
     }
+
+    private boolean isLedgerNotExistException(Integer bkErrorCode) {
+        if (bkErrorCode == null) {
+            return true;
+        }
+        switch (bkErrorCode) {
+            case BKException.Code.NoSuchLedgerExistsException:
+            case BKException.Code.NoSuchLedgerExistsOnMetadataServerException:
+                return true;
+
+            default:
+                return false;
+        }
+    }
 }