Skip replicated subscription update if the previous update was for the same snapshot id

- Reduces unnecessary updates

- Move replicated subscription advance markdelete position logic to proper code location
 - the operation is asynchronous and should be handled after the markdelete position
   updating operation completes

Author: lhotari

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

@@ -120,6 +120,8 @@ public class PersistentSubscription extends AbstractSubscription implements Subs
         REPLICATED_SUBSCRIPTION_CURSOR_PROPERTIES.put(REPLICATED_SUBSCRIPTION_PROPERTY, 1L);
     }
 
+    private volatile String lastLocalSubscriptionUpdatedSnapshotId;
+
     static Map<String, Long> getBaseCursorProperties(boolean isReplicated) {
         return isReplicated ? REPLICATED_SUBSCRIPTION_CURSOR_PROPERTIES : NON_REPLICATED_SUBSCRIPTION_CURSOR_PROPERTIES;
     }
@@ -397,21 +399,6 @@ public void acknowledgeMessage(List<Position> positions, AckType ackType, Map<St
             }
         }
 
-        if (!cursor.getMarkDeletedPosition().equals(previousMarkDeletePosition)) {
-            this.updateLastMarkDeleteAdvancedTimestamp();
-
-            // Mark delete position advance
-            ReplicatedSubscriptionSnapshotCache snapshotCache  = this.replicatedSubscriptionSnapshotCache;
-            if (snapshotCache != null) {
-                ReplicatedSubscriptionsSnapshot snapshot = snapshotCache
-                        .advancedMarkDeletePosition((PositionImpl) cursor.getMarkDeletedPosition());
-                if (snapshot != null) {
-                    topic.getReplicatedSubscriptionController()
-                            .ifPresent(c -> c.localSubscriptionUpdated(subName, snapshot));
-                }
-            }
-        }
-
         if (topic.getManagedLedger().isTerminated() && cursor.getNumberOfEntriesInBacklog(false) == 0) {
             // Notify all consumer that the end of topic was reached
             if (dispatcher != null) {
@@ -471,6 +458,26 @@ public void deleteFailed(ManagedLedgerException exception, Object ctx) {
     private void notifyTheMarkDeletePositionMoveForwardIfNeeded(Position oldPosition) {
         PositionImpl oldMD = (PositionImpl) oldPosition;
         PositionImpl newMD = (PositionImpl) cursor.getMarkDeletedPosition();
+
+        if (!newMD.equals(oldMD)) {
+            this.updateLastMarkDeleteAdvancedTimestamp();
+
+            // Mark delete position advance
+            ReplicatedSubscriptionSnapshotCache snapshotCache  = this.replicatedSubscriptionSnapshotCache;
+            if (snapshotCache != null) {
+                ReplicatedSubscriptionsSnapshot snapshot = snapshotCache.advancedMarkDeletePosition(newMD);
+                if (snapshot != null) {
+                    topic.getReplicatedSubscriptionController()
+                            .ifPresent(c -> {
+                                if (!snapshot.getSnapshotId().equals(lastLocalSubscriptionUpdatedSnapshotId)) {
+                                    c.localSubscriptionUpdated(subName, snapshot);
+                                    lastLocalSubscriptionUpdatedSnapshotId = snapshot.getSnapshotId();
+                                }
+                            });
+                }
+            }
+        }
+
         if (dispatcher != null && newMD.compareTo(oldMD) > 0) {
             dispatcher.markDeletePositionMoveForward();
         }