public void drainSources(String peerId) throws IOException, ReplicationException {
    String terminateMessage = "Sync replication peer " + peerId
      + " is transiting to STANDBY. Will close the previous replication source and open a new one";
    ReplicationPeer peer = replicationPeers.getPeer(peerId);
    assert peer.getPeerConfig().isSyncReplication();
    ReplicationQueueId queueId = new ReplicationQueueId(server.getServerName(), peerId);
    // TODO: use empty initial offsets for now, revisit when adding support for sync replication
    ReplicationSourceInterface src =
      createSource(new ReplicationQueueData(queueId, ImmutableMap.of()), peer);
    // synchronized here to avoid race with postLogRoll where we add new log to source and also
    // walsById.
    ReplicationSourceInterface toRemove;
    ReplicationQueueData queueData;
    synchronized (latestPaths) {
      // Here we make a copy of all the remaining wal files and then delete them from the
      // replication queue storage after releasing the lock. It is not safe to just remove the old
      // map from walsById since later we may fail to update the replication queue storage, and when
      // we retry next time, we can not know the wal files that needs to be set to the replication
      // queue storage
      ImmutableMap.Builder<String, ReplicationGroupOffset> builder = ImmutableMap.builder();
      synchronized (walsById) {
        walsById.get(queueId).forEach((group, wals) -> {
          if (!wals.isEmpty()) {
            builder.put(group, new ReplicationGroupOffset(wals.last(), -1));
          }
        });
      }
      queueData = new ReplicationQueueData(queueId, builder.build());
      src = createSource(queueData, peer);
      toRemove = sources.put(peerId, src);
      if (toRemove != null) {
        LOG.info("Terminate replication source for " + toRemove.getPeerId());
        toRemove.terminate(terminateMessage);
        toRemove.getSourceMetrics().clear();
      }
    }
    for (Map.Entry<String, ReplicationGroupOffset> entry : queueData.getOffsets().entrySet()) {
      queueStorage.setOffset(queueId, entry.getKey(), entry.getValue(), Collections.emptyMap());
    }
    LOG.info("Startup replication source for " + src.getPeerId());
    src.startup();
    synchronized (walsById) {
      Map<String, NavigableSet<String>> wals = walsById.get(queueId);
      queueData.getOffsets().forEach((group, offset) -> {
        NavigableSet<String> walsByGroup = wals.get(group);
        if (walsByGroup != null) {
          walsByGroup.headSet(offset.getWal(), true).clear();
        }
      });
    }
    // synchronized on oldsources to avoid race with NodeFailoverWorker. Since NodeFailoverWorker is
    // a background task, we will delete the file from replication queue storage under the lock to
    // simplify the logic.
    synchronized (this.oldsources) {
      for (Iterator<ReplicationSourceInterface> iter = oldsources.iterator(); iter.hasNext();) {
        ReplicationSourceInterface oldSource = iter.next();
        if (oldSource.getPeerId().equals(peerId)) {
          ReplicationQueueId oldSourceQueueId = oldSource.getQueueId();
          oldSource.terminate(terminateMessage);
          oldSource.getSourceMetrics().clear();
          queueStorage.removeQueue(oldSourceQueueId);
          walsByIdRecoveredQueues.remove(oldSourceQueueId);
          iter.remove();
        }
      }
    }
  }
<p>
This is used when we transit a sync replication peer to {@link SyncReplicationState#STANDBY}.
</p>
<p>
When transiting to {@link SyncReplicationState#STANDBY}, we can remove all the pending wal
files for a replication peer as we do not need to replicate them any more. And this is
necessary, otherwise when we transit back to {@link SyncReplicationState#DOWNGRADE_ACTIVE}
later, the stale data will be replicated again and cause inconsistency.
</p>
<p>
See HBASE-20426 for more details.
</p>
@param peerId the id of the sync replication peer
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java