[GitHub] [pulsar] massakam commented on a change in pull request #6592: [Issue 6437][broker] Prevent marker messages from accumulating in backlog of replicated subscription

[GitBox <gi...@apache.org>]

massakam commented on a change in pull request #6592:
URL: https://github.com/apache/pulsar/pull/6592#discussion_r399173100



##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsSnapshotBuilder.java
##########
@@ -122,9 +123,18 @@ synchronized void receivedSnapshotResponse(Position position, ReplicatedSubscrip
         }
         // Snapshot is now complete, store it in the local topic
         PositionImpl p = (PositionImpl) position;
-        controller.writeMarker(
-                Markers.newReplicatedSubscriptionsSnapshot(snapshotId, controller.localCluster(),
-                        p.getLedgerId(), p.getEntryId(), responses));
+        try {
+            ReplicatedSubscriptionsSnapshot snapshot = Markers.instantiateReplicatedSubscriptionsSnapshot(snapshotId,
+                    controller.localCluster(), p.getLedgerId(), p.getEntryId(), responses);
+            controller.topic().getSubscriptions().forEach((subName, sub) -> {
+                if (sub != null) {
+                    sub.processReplicatedSubscriptionSnapshot(snapshot);
+                }
+            });
+        } catch (Throwable t) {
+            log.warn("[{}] Failed to process replicated subscription snapshot {} -- {}", controller.topic().getName(),
+                    snapshotId, t.getMessage());
+        }

Review comment:
       An instance of `ReplicatedSubscriptionSnapshotCache` should hold up to 10 snapshots. 
   https://github.com/apache/pulsar/blob/7be1ee1fdb59421ac858b38840d3baf8c9073a5c/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionSnapshotCache.java#L45-L60
   If we publish `ReplicatedSubscriptionsSnapshot` messages to a local topic and acknowledge only the messages that have been deleted from the cache, can we avoid the issue you are worried about?




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org