[Broker] Fix replicated subscriptions direct memory leak (#11396)

- .release() wasn't called for marker messages sent by ReplicatedSubscriptionsController.
- Use writeMarker method to publish marker messages and call release in try-finally block.

(cherry picked from commit 8841c81)

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

@@ -117,7 +117,7 @@ public void localSubscriptionUpdated(String subscriptionName, ReplicatedSubscrip
         }
 
         ByteBuf subscriptionUpdate = Markers.newReplicatedSubscriptionsUpdate(subscriptionName, clusterIds);
-        topic.publishMessage(subscriptionUpdate, this);
+        writeMarker(subscriptionUpdate);
     }
 
     private void receivedSnapshotRequest(ReplicatedSubscriptionsSnapshotRequest request) {
@@ -134,8 +134,7 @@ private void receivedSnapshotRequest(ReplicatedSubscriptionsSnapshotRequest requ
                 request.getSourceCluster(),
                 localCluster,
                 lastMsgId.getLedgerId(), lastMsgId.getEntryId());
-
-        topic.publishMessage(marker, this);
+        writeMarker(marker);
     }
 
     private void receivedSnapshotResponse(Position position, ReplicatedSubscriptionsSnapshotResponse response) {
@@ -228,7 +227,11 @@ void snapshotCompleted(String snapshotId) {
     }
 
     void writeMarker(ByteBuf marker) {
-        topic.publishMessage(marker, this);
+        try {
+            topic.publishMessage(marker, this);
+        } finally {
+            marker.release();
+        }
     }
 
     /**