public void addMetricsForRemoteReplicaInfo(RemoteReplicaInfo remoteReplicaInfo, boolean trackPerDatacenterLag,
      boolean trackPerReplicaReplicationBytes) {
    String metricNamePrefix = generateRemoteReplicaMetricPrefix(remoteReplicaInfo);

    String metadataRequestErrorMetricName = metricNamePrefix + "-metadataRequestError";
    if (metadataRequestErrorMap.containsKey(metadataRequestErrorMetricName)) {
      // Metrics already exist. For VCR: Partition add/remove go back and forth.
      return;
    }
    Counter metadataRequestError =
        registry.counter(MetricRegistry.name(ReplicaThread.class, metadataRequestErrorMetricName));
    metadataRequestErrorMap.put(metadataRequestErrorMetricName, metadataRequestError);

    String getRequestErrorMetricName = metricNamePrefix + "-getRequestError";
    Counter getRequestError = registry.counter(MetricRegistry.name(ReplicaThread.class, getRequestErrorMetricName));
    getRequestErrorMap.put(getRequestErrorMetricName, getRequestError);

    String localStoreErrorMetricName = metricNamePrefix + "-localStoreError";
    Counter localStoreError = registry.counter(MetricRegistry.name(ReplicaThread.class, localStoreErrorMetricName));
    localStoreErrorMap.put(localStoreErrorMetricName, localStoreError);

    if (trackPerReplicaReplicationBytes) {
      String replicaToReplicationFetchBytesRateMetricName = metricNamePrefix + REPLICATION_FETCH_BYTES_RATE_SUFFIX;
      Meter replicaToReplicationFetchBytesRate =
          registry.meter(MetricRegistry.name(ReplicaThread.class, replicaToReplicationFetchBytesRateMetricName));
      replicaToReplicationFetchBytesRateMap.put(replicaToReplicationFetchBytesRateMetricName,
          replicaToReplicationFetchBytesRate);
    }

    Gauge<Long> replicaLag = remoteReplicaInfo::getRemoteLagFromLocalInBytes;
    registry.gauge(MetricRegistry.name(ReplicaThread.class, metricNamePrefix + "-remoteLagInBytes"), () -> replicaLag);
    if (trackPerDatacenterLag) {
      String remoteReplicaDc = remoteReplicaInfo.getReplicaId().getDataNodeId().getDatacenterName();
      remoteReplicaInfosByDc.computeIfAbsent(remoteReplicaDc, k -> {
        Gauge<Double> avgReplicaLag = () -> getAvgLagFromDc(remoteReplicaDc);
        registry.gauge(MetricRegistry.name(ReplicaThread.class, remoteReplicaDc + "-avgReplicaLagFromLocalInBytes"),
            () -> avgReplicaLag);
        Gauge<Long> maxReplicaLag = () -> {
          LongSummaryStatistics statistics =
              dcToReplicaLagStats.getOrDefault(remoteReplicaDc, new LongSummaryStatistics());
          return statistics.getMax() == Long.MIN_VALUE ? -1 : statistics.getMax();
        };
        registry.gauge(MetricRegistry.name(ReplicaThread.class, remoteReplicaDc + "-maxReplicaLagFromLocalInBytes"),
            () -> maxReplicaLag);
        Gauge<Long> minReplicaLag = () -> {
          LongSummaryStatistics statistics =
              dcToReplicaLagStats.getOrDefault(remoteReplicaDc, new LongSummaryStatistics());
          return statistics.getMin() == Long.MAX_VALUE ? -1 : statistics.getMin();
        };
        registry.gauge(MetricRegistry.name(ReplicaThread.class, remoteReplicaDc + "-minReplicaLagFromLocalInBytes"),
            () -> minReplicaLag);
        return ConcurrentHashMap.newKeySet();
      }).add(remoteReplicaInfo);
    }
  }
Add metrics for given remote replica.
@param remoteReplicaInfo the {@link RemoteReplicaInfo} that contains all infos associated with remote replica
@param trackPerDatacenterLag whether to track remote replicas' lag from local by each datacenter. Specifically, it
                             tracks min/max/avg remote replica lag from each datacenter.
@param trackPerReplicaReplicationBytes whether to track replicated bytes per second from each remote replica,
                                       replica thread is replicating from.
ambry-replication/src/main/java/com/github/ambry/replication/ReplicationMetrics.java