docs: void handleFragmentTaskMessage(FragmentTaskMessage message)
      {
          FragmentTaskMessage msg = message;
          long newSpHandle;
          //The site has been marked as non-leader. The follow-up batches or fragments are processed here
          if (!message.isForReplica() && (m_isLeader || message.isExecutedOnPreviousLeader())) {
              // message processed on leader
              // Quick hack to make progress...we need to copy the FragmentTaskMessage
              // before we start mucking with its state (SPHANDLE).  We need to revisit
              // all the messaging mess at some point.
              msg = new FragmentTaskMessage(message.getInitiatorHSId(),
                      message.getCoordinatorHSId(), message);
              //Not going to use the timestamp from the new Ego because the multi-part timestamp is what should be used
              msg.setTimestamp(message.getTimestamp());
              msg.setExecutedOnPreviousLeader(message.isExecutedOnPreviousLeader());
              if (!message.isReadOnly()) {
                  TxnEgo ego = advanceTxnEgo();
                  newSpHandle = ego.getTxnId();
  
                  if (m_outstandingTxns.get(msg.getTxnId()) == null) {
                      updateMaxScheduledTransactionSpHandle(newSpHandle);
                  }
              } else {
                  newSpHandle = getMaxScheduledTxnSpHandle();
              }
  
              msg.setSpHandle(newSpHandle);
              msg.setLastSpUniqueId(m_uniqueIdGenerator.getLastUniqueId());
              logRepair(msg);
              if (msg.getInitiateTask() != null) {
                  msg.getInitiateTask().setSpHandle(newSpHandle);//set the handle
                  //Trigger reserialization so the new handle is used
                  msg.setStateForDurability(msg.getInitiateTask(), msg.getInvolvedPartitions());
              }
  
              /*
               * If there a replicas to send it to, forward it!
               * Unless... it's read only AND not a sysproc. Read only sysprocs may expect to be sent
               * everywhere.
               * In that case don't propagate it to avoid a determinism check and extra messaging overhead
               */
              if (IS_KSAFE_CLUSTER && (!message.isReadOnly() || msg.isSysProcTask())) {
                  for (long hsId : m_sendToHSIds) {
                      FragmentTaskMessage finalMsg = msg;
                      final VoltTrace.TraceEventBatch traceLog = VoltTrace.log(VoltTrace.Category.SPI);
                      if (traceLog != null) {
                          traceLog.add(() -> VoltTrace.beginAsync("replicatefragment",
                                                                  MiscUtils.hsIdPairTxnIdToString(m_mailbox.getHSId(), hsId, finalMsg.getSpHandle(), finalMsg.getTxnId()),
                                                                  "txnId", TxnEgo.txnIdToString(finalMsg.getTxnId()),
                                                                  "dest", CoreUtils.hsIdToString(hsId)));
                      }
                  }
  
                  FragmentTaskMessage replmsg =
                      new FragmentTaskMessage(m_mailbox.getHSId(),
                              m_mailbox.getHSId(), msg);
                  replmsg.setForReplica(true);
                  replmsg.setTimestamp(msg.getTimestamp());
                  // K-safety cluster doesn't always mean partition has replicas,
                  // node failure may reduce the number of replicas for each partition.
                  if (m_sendToHSIds.length > 0) {
                      m_mailbox.send(m_sendToHSIds,replmsg);
                  }
                  DuplicateCounter counter;
                  /*
                   * Non-determinism should be impossible to happen with MP fragments.
                   * if you see "MP_DETERMINISM_ERROR" as procedure name in the crash logs
                   * something has horribly gone wrong.
                   */
                  if (message.getFragmentTaskType() != FragmentTaskMessage.SYS_PROC_PER_SITE) {
                      counter = new DuplicateCounter(
                              msg.getCoordinatorHSId(),
                              msg.getTxnId(),
                              m_replicaHSIds,
                              replmsg);
                  }
                  else {
                      counter = new SysProcDuplicateCounter(
                              msg.getCoordinatorHSId(),
                              msg.getTxnId(),
                              m_replicaHSIds,
                              replmsg);
                  }
                  safeAddToDuplicateCounterMap(new DuplicateCounterKey(message.getTxnId(), newSpHandle), counter);
              }
          } else {
              // message processed on replica
              logRepair(msg);
              newSpHandle = msg.getSpHandle();
              setMaxSeenTxnId(newSpHandle);
          }
  
          Iv2Trace.logFragmentTaskMessage(message, m_mailbox.getHSId(), newSpHandle, false);
          doLocalFragmentOffer(msg);
      }