[pulsar-broker] configure maxMsgReplDelayInSeconds for each colo (apa…

…che#8409)

### Motivation
`maxMsgReplDelayInSeconds` is useful metrics for replicator n/w monitoring and we should have it in namespace-metrics for each colo. We have added max delay across all replication-clusters in apache#2983 but it will be more useful and make-sense to add at each colo for monitoring purpose.

### Result
After fix it should appear in namespace-replicator stats
```
./pulsar-admin broker-stats monitoring-metrics -i

{
    "metrics": {
        "brk_max_replication_delay_second": 0.0,
        "brk_repl_is_connected": 1,
        "brk_repl_out_rate": 0.03857322653872893,
        "brk_repl_out_tp_rate": 0.23143935923237358,
        "brk_replication_backlog": 0.0
    },
    "dimensions": {
        "from_cluster": "r1",
        "namespace": "pulsar/ns",
        "to_cluster": "r3"
    }
}
```

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

@@ -1404,10 +1404,6 @@ public void updateRates(NamespaceStats nsStats, NamespaceBundleStats bundleStats
             topicStatsStream.endObject();
 
             nsStats.msgReplBacklog += rStat.replicationBacklog;
-            // replication delay for a namespace is the max repl-delay among all the topics under this namespace
-            if (rStat.replicationDelayInSeconds > nsStats.maxMsgReplDelayInSeconds) {
-                nsStats.maxMsgReplDelayInSeconds = rStat.replicationDelayInSeconds;
-            }
 
             if (replStats.isMetricsEnabled()) {
                 String namespaceClusterKey = replStats.getKeyName(namespace, cluster);
@@ -1424,6 +1420,10 @@ public void updateRates(NamespaceStats nsStats, NamespaceBundleStats bundleStats
                 if (update) {
                     replStats.put(namespaceClusterKey, replicationMetrics);
                 }
+                // replication delay for a namespace is the max repl-delay among all the topics under this namespace
+                if (rStat.replicationDelayInSeconds > replicationMetrics.maxMsgReplDelayInSeconds) {
+                    replicationMetrics.maxMsgReplDelayInSeconds = rStat.replicationDelayInSeconds;
+                }
             }
         });
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/ReplicationMetrics.java

@@ -33,13 +33,15 @@ public class ReplicationMetrics {
     public double msgRateOut;
     public double msgThroughputOut;
     public double msgReplBacklog;
+    public double maxMsgReplDelayInSeconds;
     public int connected;
 
     public void reset() {
         msgRateOut = 0;
         msgThroughputOut = 0;
         msgReplBacklog = 0;
         connected = 0;
+        maxMsgReplDelayInSeconds = 0;
     }
 
     public static ReplicationMetrics get() {
@@ -81,6 +83,7 @@ public Metrics add(String namespace, String local, String remote) {
         dMetrics.put("brk_repl_out_tp_rate", msgThroughputOut);
         dMetrics.put("brk_replication_backlog", msgReplBacklog);
         dMetrics.put("brk_repl_is_connected", connected);
+        dMetrics.put("brk_max_replication_delay_second", maxMsgReplDelayInSeconds);
 
         return dMetrics;