[improve][broker]Skip to mark delete if the target position of expira… (apache#24881)

Co-authored-by: Jiwei Guo <[email protected]>
Co-authored-by: Zixuan Liu <[email protected]>
(cherry picked from commit 14b0821)
(cherry picked from commit 5bdc5aa)

Author: 3 people

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

@@ -232,6 +232,10 @@ public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
     @Override
     public void findEntryComplete(Position position, Object ctx) {
         if (position != null) {
+            var markDeletedPosition = cursor.getMarkDeletedPosition();
+            if (markDeletedPosition != null && markDeletedPosition.compareTo(position) >= 0) {
+                return;
+            }
             log.info("[{}][{}] Expiring all messages until position {}", topicName, subName, position);
             Position prevMarkDeletePos = cursor.getMarkDeletedPosition();
             cursor.asyncMarkDelete(position, cursor.getProperties(), markDeleteCallback,