[Broker] Fix replicated subscriptions related LightProto issues (#10247)

* [Broker] Make copy of MarkersMessageIdData since instances are reused

- LightProto reuses thread local instances and therefore a copy must be made

* [Broker] Make copy of ReplicatedSubscriptionsSnapshot since instances are reused

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

@@ -1059,7 +1059,7 @@ protected Map<String, Long> mergeCursorProperties(Map<String, Long> userProperti
     public void processReplicatedSubscriptionSnapshot(ReplicatedSubscriptionsSnapshot snapshot) {
         ReplicatedSubscriptionSnapshotCache snapshotCache = this.replicatedSubscriptionSnapshotCache;
         if (snapshotCache != null) {
-            snapshotCache.addNewSnapshot(snapshot);
+            snapshotCache.addNewSnapshot(new ReplicatedSubscriptionsSnapshot().copyFrom(snapshot));
         }
     }
 

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

@@ -90,7 +90,7 @@ synchronized void receivedSnapshotResponse(Position position, ReplicatedSubscrip
                     response.getCluster().getCluster());
         }
         String cluster = response.getCluster().getCluster();
-        responses.putIfAbsent(cluster, response.getCluster().getMessageId());
+        responses.putIfAbsent(cluster, new MarkersMessageIdData().copyFrom(response.getCluster().getMessageId()));
         missingClusters.remove(cluster);
 
         if (log.isDebugEnabled()) {