@Test
  public void dcLevelReplicationLagMetricsTest() throws Exception {
    MockClusterMap clusterMap = new MockClusterMap();
    ClusterMapConfig clusterMapConfig = new ClusterMapConfig(verifiableProperties);
    Pair<StorageManager, ReplicationManager> managers =
        createStorageManagerAndReplicationManager(clusterMap, clusterMapConfig, null);
    MockReplicationManager replicationManager = (MockReplicationManager) managers.getSecond();
    Set<String> remoteDcNames = new HashSet<>(Arrays.asList("DC1", "DC2", "DC3"));
    String localDcName = clusterMap.getDataNodeIds().get(0).getDatacenterName();
    remoteDcNames.remove(localDcName);

    // before updating replication lag, make sure avg lag in each dc is 0
    MetricRegistry metricRegistry = replicationManager.getMetricRegistry();
    String prefix = ReplicaThread.class.getName() + ".";
    String avgMetricSuffix = "-avgReplicaLagFromLocalInBytes";
    assertEquals("Average replication lag in local dc is not expected", 18.0,
        metricRegistry.getGauges().get(prefix + localDcName + avgMetricSuffix).getValue());
    for (String remoteDc : remoteDcNames) {
      assertEquals("Average replication lag in remote dc is not expected", 18.0,
          metricRegistry.getGauges().get(prefix + remoteDc + avgMetricSuffix).getValue());
    }

    // iterate over all partitions on current node and make sure all their peer replicas in local dc have fully caught up
    for (Map.Entry<PartitionId, PartitionInfo> entry : replicationManager.partitionToPartitionInfo.entrySet()) {
      PartitionId localPartition = entry.getKey();
      PartitionInfo partitionInfo = entry.getValue();
      List<RemoteReplicaInfo> remoteReplicaInfos = partitionInfo.getRemoteReplicaInfos()
          .stream()
          .filter(info -> info.getReplicaId().getDataNodeId().getDatacenterName().equals(localDcName))
          .collect(Collectors.toList());
      for (RemoteReplicaInfo remoteReplicaInfoInLocalDc : remoteReplicaInfos) {
        ReplicaId peerReplicaInLocalDc = remoteReplicaInfoInLocalDc.getReplicaId();
        replicationManager.updateTotalBytesReadByRemoteReplica(localPartition,
            peerReplicaInLocalDc.getDataNodeId().getHostname(), peerReplicaInLocalDc.getReplicaPath(), 18);
      }
    }
    // verify that after updating replication lag for all peer replicas in local dc, the avg lag in local dc has updated
    assertEquals("Average replication lag in local dc is not expected", 0.0,
        metricRegistry.getGauges().get(prefix + localDcName + avgMetricSuffix).getValue());
    // for remote dc, the avg lag is still 18.0
    for (String remoteDc : remoteDcNames) {
      assertEquals("Average replication lag in remote dc is not expected", 18.0,
          metricRegistry.getGauges().get(prefix + remoteDc + avgMetricSuffix).getValue());
    }
  }
Test that metrics that track remote replicas lag behind local replicas in each dc.
@throws Exception
ambry-replication/src/test/java/com/github/ambry/replication/ReplicationTest.java