docs: private void doLocalFragmentOffer(FragmentTaskMessage msg)
      {
          final String threadName = Thread.currentThread().getName(); // Thread name has to be materialized here
          final VoltTrace.TraceEventBatch traceLog = VoltTrace.log(VoltTrace.Category.SPI);
          if (traceLog != null) {
              traceLog.add(() -> VoltTrace.meta("process_name", "name", CoreUtils.getHostnameOrAddress()))
                      .add(() -> VoltTrace.meta("thread_name", "name", threadName))
                      .add(() -> VoltTrace.meta("thread_sort_index", "sort_index", Integer.toString(10000)))
                      .add(() -> VoltTrace.beginAsync("recvfragment",
                                                      MiscUtils.hsIdPairTxnIdToString(m_mailbox.getHSId(), m_mailbox.getHSId(), msg.getSpHandle(), msg.getTxnId()),
                                                      "txnId", TxnEgo.txnIdToString(msg.getTxnId()),
                                                      "partition", m_partitionId,
                                                      "hsId", CoreUtils.hsIdToString(m_mailbox.getHSId()),
                                                      "final", msg.isFinalTask()));
          }
  
          TransactionState txn = m_outstandingTxns.get(msg.getTxnId());
          boolean logThis = false;
          // bit of a hack...we will probably not want to create and
          // offer FragmentTasks for txn ids that don't match if we have
          // something in progress already
          if (txn == null) {
              //FIXME: replay transaction doesn't have initiate task,
              txn = new ParticipantTransactionState(msg.getSpHandle(), msg, msg.isReadOnly(),
                      msg.isNPartTxn());
              m_outstandingTxns.put(msg.getTxnId(), txn);
              // Only want to send things to the command log if it satisfies this predicate
              // AND we've never seen anything for this transaction before.  We can't
              // actually log until we create a TransactionTask, though, so just keep track
              // of whether it needs to be done.
  
              // Like SP, we should log writes and safe reads.
              logThis = true;
          }
  
          // Check to see if this is the final task for this txn, and if so, if we can close it out early
          // Right now, this just means read-only.
          // NOTE: this overlaps slightly with CompleteTransactionMessage handling completion.  It's so tiny
          // that for now, meh, but if this scope grows then it should get refactored out
          if (msg.isFinalTask() && txn.isReadOnly()) {
              m_outstandingTxns.remove(msg.getTxnId());
          }
  
          TransactionTask task;
          if (msg.isSysProcTask()) {
              task =
                  new SysprocFragmentTask(m_mailbox, (ParticipantTransactionState)txn,
                                          m_pendingTasks, msg, null);
          }
          else {
              task =
                  new FragmentTask(m_mailbox, (ParticipantTransactionState)txn,
                                   m_pendingTasks, msg, null);
          }
          if (logThis) {
              ListenableFuture<Object> durabilityBackpressureFuture =
                      m_cl.log(msg.getInitiateTask(), msg.getSpHandle(), Ints.toArray(msg.getInvolvedPartitions()),
                               m_durabilityListener, task);
  
              if (traceLog != null && durabilityBackpressureFuture != null) {
                  traceLog.add(() -> VoltTrace.beginAsync("durability",
                                                          MiscUtils.hsIdTxnIdToString(m_mailbox.getHSId(), msg.getSpHandle()),
                                                          "txnId", TxnEgo.txnIdToString(msg.getTxnId()),
                                                          "partition", Integer.toString(m_partitionId)));
              }
  
              //Durability future is always null for sync command logging
              //the transaction will be delivered again by the CL for execution once durable
              //Async command logging has to offer the task immediately with a Future for backpressure
              if (m_cl.canOfferTask()) {
                  m_pendingTasks.offer(task.setDurabilityBackpressureFuture(durabilityBackpressureFuture));
              } else {
                  /* Getting here means that the task is the first fragment of an MP txn and
                   * synchronous command logging is on, so create a backlog for future tasks of
                   * this MP arrived before it's marked durable.
                   *
                   * This is important for synchronous command logging and MP txn restart. Without
                   * this, a restarted MP txn may not be gated by logging of the first fragment.
                   */
                  assert !m_mpsPendingDurability.containsKey(task.getTxnId());
                  m_mpsPendingDurability.put(task.getTxnId(), new ArrayDeque<TransactionTask>());
              }
          } else {
              queueOrOfferMPTask(task);
          }
      }