@Override
  public void updateTotalBytesReadByRemoteReplica(PartitionId partitionId, String hostName, String replicaPath,
      long totalBytesRead) throws StoreException {
    RemoteReplicaInfo remoteReplicaInfo = getRemoteReplicaInfo(partitionId, hostName, replicaPath);
    if (remoteReplicaInfo != null) {
      ReplicaId localReplica = remoteReplicaInfo.getLocalReplicaId();
      remoteReplicaInfo.setTotalBytesReadFromLocalStore(totalBytesRead);
      // update replication lag in ReplicaSyncUpManager
      if (replicaSyncUpManager != null) {
        Store localStore = storeManager.getStore(partitionId);
        if (localStore.getCurrentState() == ReplicaState.INACTIVE) {
          // if local store is in INACTIVE state, that means deactivation process is initiated and in progress on this
          // replica. We update SyncUpManager by peer's lag from last PUT offset in local store.
          // it's ok if deactivation has completed and concurrent metadata request attempts to update lag of same replica
          // again. The reason is, SyncUpManager has a lock to ensure only one request will call onDeactivationComplete()
          // method. The local replica should have been removed when another request acquires the lock.
          replicaSyncUpManager.updateReplicaLagAndCheckSyncStatus(localReplica, remoteReplicaInfo.getReplicaId(),
              localStore.getEndPositionOfLastPut() - totalBytesRead, ReplicaState.INACTIVE);
        } else if (localStore.getCurrentState() == ReplicaState.OFFLINE && localStore.isDecommissionInProgress()) {
          // if local store is in OFFLINE state, we need more info to determine if replica is really in Inactive-To-Offline
          // transition. So we check if decommission file is present. If present, we update SyncUpManager by peer's lag
          // from end offset in local store.
          replicaSyncUpManager.updateReplicaLagAndCheckSyncStatus(localReplica, remoteReplicaInfo.getReplicaId(),
              localStore.getSizeInBytes() - totalBytesRead, ReplicaState.OFFLINE);
        }
      }
    }
  }
Enable/Disable replication for given partitions if all origins are involved,
we can disable replication for all partitions and pass this to future ReplicaThreads as well
otherwise control replication for current replica threads only
@param ids the {@link PartitionId}s to enable/disable it on.
@param origins the list of datacenters from which replication should be enabled/disabled. Having an empty list
               disables replication from all datacenters.
@param enable whether to enable ({@code true}) or disable.
@return true/false
ambry-replication/src/main/java/com/github/ambry/replication/ReplicationEngine.java