docs: @Override
      public long[] updateReplicas(List<Long> replicas, Map<Integer, Long> partitionMasters,
              TransactionState snapshotTransactionState)
      {
          if (tmLog.isDebugEnabled()) {
              tmLog.debug("[SpScheduler.updateReplicas] replicas to " + CoreUtils.hsIdCollectionToString(replicas) +
                      " on " + CoreUtils.hsIdToString(m_mailbox.getHSId())
               + " from " + CoreUtils.hsIdCollectionToString(m_replicaHSIds));
          }
          long[] replicasAdded = new long[0];
          if (m_replicaHSIds.size() > 0 && replicas.size() > m_replicaHSIds.size()) {
              // Remember the rejoin sites before update replicas set
              Set<Long> rejoinHSIds = Sets.difference(new HashSet<Long>(replicas),
                                                    new HashSet<Long>(m_replicaHSIds));
              replicasAdded = Longs.toArray(rejoinHSIds);
          }
          // First - correct the official replica set.
          m_replicaHSIds = replicas;
          // Update the list of remote replicas that we'll need to send to
          List<Long> sendToHSIds = new ArrayList<Long>(m_replicaHSIds);
          sendToHSIds.remove(m_mailbox.getHSId());
          m_sendToHSIds = Longs.toArray(sendToHSIds);
  
          // A new site joins in, forward the current txn (stream snapshot save) message to new site
          if (m_isLeader && snapshotTransactionState != null) {
              // Look up the DuplicateCounter for this snapshots fragment
              DuplicateCounterKey key = new DuplicateCounterKey(snapshotTransactionState.txnId,
                      snapshotTransactionState.m_spHandle);
              DuplicateCounter duplicateCounter = m_duplicateCounters.get(key);
              assert (duplicateCounter != null);
              duplicateCounter.addReplicas(replicasAdded);
              // Forward fragment message to new replica
              m_mailbox.send(replicasAdded, duplicateCounter.getOpenMessage());
          }
  
          // Cleanup duplicate counters and collect DONE counters
          // in this list for further processing.
          List<DuplicateCounterKey> doneCounters = new LinkedList<DuplicateCounterKey>();
          for (Entry<DuplicateCounterKey, DuplicateCounter> entry : m_duplicateCounters.entrySet()) {
              DuplicateCounter counter = entry.getValue();
              int result = counter.updateReplicas(m_replicaHSIds);
              if (result == DuplicateCounter.DONE) {
                  doneCounters.add(entry.getKey());
              }
          }
  
          //notify the new partition leader that the old leader has completed the Txns if needed
          //after duplicate counters are cleaned. m_mailbox can be MockMailBox which is used for
          //unit test
          if (!m_isLeader && m_mailbox instanceof InitiatorMailbox) {
              ((InitiatorMailbox)m_mailbox).notifyNewLeaderOfTxnDoneIfNeeded();
          }
  
          // Maintain the CI invariant that responses arrive in txnid order.
          Collections.sort(doneCounters);
          for (DuplicateCounterKey key : doneCounters) {
              DuplicateCounter counter = m_duplicateCounters.remove(key);
  
              final TransactionState txn = m_outstandingTxns.get(key.m_txnId);
              if (txn == null || txn.isDone()) {
                  m_outstandingTxns.remove(key.m_txnId);
                  // for MP write txns, we should use it's first SpHandle in the TransactionState
                  // for SP write txns, we can just use the SpHandle from the DuplicateCounterKey
                  long safeSpHandle = txn == null ? key.m_spHandle: txn.m_spHandle;
                  setRepairLogTruncationHandle(safeSpHandle, false);
              }
  
              VoltMessage resp = counter.getLastResponse();
              if (resp != null) {
                  // MPI is tracking deps per partition HSID.  We need to make
                  // sure we write ours into the message getting sent to the MPI
                  if (resp instanceof FragmentResponseMessage) {
                      FragmentResponseMessage fresp = (FragmentResponseMessage)resp;
                      fresp.setExecutorSiteId(m_mailbox.getHSId());
                  }
                  m_mailbox.send(counter.m_destinationId, resp);
              }
              else {
                  hostLog.warn("TXN " + counter.getTxnId() + " lost all replicas and " +
                          "had no responses.  This should be impossible?");
              }
          }
  
          long uniqueId;
          if (snapshotTransactionState == null) {
              uniqueId = m_uniqueIdGenerator.getLastUniqueId();
          } else {
              // When there is a snapshot transaction state use the lastSpUniqueId from that
              uniqueId = snapshotTransactionState.m_lastSpUniqueId;
          }
          SettableFuture<Boolean> written = writeIv2ViableReplayEntry(uniqueId);
  
          // Get the fault log status here to ensure the leader has written it to disk
          // before initiating transactions again.
          blockFaultLogWriteStatus(written);
  
          return replicasAdded;
      }