docs: public VoltTable startSnapshotting(
              final String file_path, final String pathType, final String file_nonce, final SnapshotFormat format, final byte block,
              final long multiPartTxnId, final long partitionTxnId, final long legacyPerPartitionTxnIds[],
              final String data, final SystemProcedureExecutionContext context, final String hostname,
              final HashinatorSnapshotData hashinatorData,
              final long timestamp)
      {
          // TRAIL [SnapSave:3] 3 [all SP] Start to execute create snapshot target requests.
          TRACE_LOG.trace("Creating snapshot target and handing to EEs");
          final VoltTable result = SnapshotUtil.constructNodeResultsTable();
          JSONObject jsData = null;
          if (data != null && !data.isEmpty()) {
              try {
                  jsData = new JSONObject(data);
              }
              catch (JSONException e) {
                  SNAP_LOG.error(String.format("JSON exception on snapshot data \"%s\".", data),
                          e);
              }
          }
          final JSONObject finalJsData = jsData;
  
          JSONObject perSiteRemoteDataCenterDrIds;
          try {
              perSiteRemoteDataCenterDrIds = ExtensibleSnapshotDigestData.serializeSiteConsumerDrIdTrackersToJSON(
                      context.getDrAppliedTrackers());
          }
          catch (JSONException e) {
              SNAP_LOG.warn("Failed to serialize the Remote DataCenter's Last applied DRIds");
              perSiteRemoteDataCenterDrIds = new JSONObject();
          }
  
          // One site wins the race to create the snapshot targets, populating
          // m_taskListsForHSIds for the other sites and creating an appropriate
          // number of snapshot permits.
          synchronized (SnapshotSiteProcessor.m_snapshotCreateLock) {
  
              SnapshotSiteProcessor.m_snapshotCreateSetupBarrierActualAction.set(new Runnable() {
                  @Override
                  public void run() {
                      Map<Integer, Long> partitionTransactionIds = m_partitionLastSeenTransactionIds;
  
                      SNAP_LOG.debug("Last seen partition transaction ids " + partitionTransactionIds);
                      m_partitionLastSeenTransactionIds = new HashMap<Integer, Long>();
                      partitionTransactionIds.put(TxnEgo.getPartitionId(multiPartTxnId), multiPartTxnId);
  
                      Map<Integer, JSONObject> remoteDataCenterLastIds = m_remoteDataCenterLastIds;
                      m_remoteDataCenterLastIds = new HashMap<Integer, JSONObject>();
  
                      /*
                       * Do a quick sanity check that the provided IDs
                       * don't conflict with currently active partitions. If they do
                       * it isn't fatal we can just skip it.
                       */
                      for (long txnId : legacyPerPartitionTxnIds) {
                          final int legacyPartition = TxnEgo.getPartitionId(txnId);
                          if (partitionTransactionIds.containsKey(legacyPartition)) {
                              SNAP_LOG.warn("While saving a snapshot and propagating legacy " +
                                  "transaction ids found an id that matches currently active partition" +
                                  partitionTransactionIds.get(legacyPartition));
                          } else {
                              partitionTransactionIds.put( legacyPartition, txnId);
                          }
                      }
  
                      ElasticService es = VoltDB.instance().getElasticService();
                      m_allLocalSiteSnapshotDigestData = new ExtensibleSnapshotDigestData(
                              SnapshotSiteProcessor.getExportSequenceNumbers(),
                              SnapshotSiteProcessor.getDRTupleStreamStateInfo(),
                              remoteDataCenterLastIds, es == null ? null : es.getResumeMetadata(),
                              finalJsData);
                      m_allLocalSiteSnapshotDigestData.setDisabledExternalStreams(
                              SnapshotSiteProcessor.getDisabledExternalStreams());
                      createSetupIv2(
                              file_path,
                              pathType,
                              file_nonce,
                              format,
                              multiPartTxnId,
                              partitionTransactionIds,
                              finalJsData,
                              context,
                              result,
                              m_allLocalSiteSnapshotDigestData,
                              context.getSiteTrackerForSnapshot(),
                              hashinatorData,
                              timestamp);
                  }
              });
  
              // Create a barrier to use with the current number of sites to wait for
              // or if the barrier is already set up check if it is broken and reset if necessary
              final int numLocalSites = context.getLocalSitesCount();
              SnapshotSiteProcessor.readySnapshotSetupBarriers(numLocalSites);
  
              //From within this EE, record the sequence numbers as of the start of the snapshot (now)
              //so that the info can be put in the digest.
              SnapshotSiteProcessor.populateExternalStreamsStatesFromSites(context);
              Integer partitionId = TxnEgo.getPartitionId(partitionTxnId);
              if (SNAP_LOG.isDebugEnabled()) {
                  SNAP_LOG.debug("Registering transaction id " + partitionTxnId + " for " + TxnEgo.getPartitionId(partitionTxnId) + " SP Txn:" +
                          TxnEgo.txnIdSeqToString(partitionTxnId) + " MP Txn:" + TxnEgo.txnIdSeqToString(multiPartTxnId));
              }
              m_partitionLastSeenTransactionIds.put(partitionId, partitionTxnId);
              m_remoteDataCenterLastIds.put(partitionId, perSiteRemoteDataCenterDrIds);
              m_isTruncation = finalJsData != null && finalJsData.has("truncReqId");
          }
  
          boolean runPostTasks = false;
          VoltTable earlyResultTable = null;
          try {
              //if all sites can not come together in 30 seconds, some sites may never receive
              //request to start snapshot, most likely some nodes are down. So break out the barrier
              //for early exit.
              try {
                  SnapshotSiteProcessor.m_snapshotCreateSetupBarrier.await(30, TimeUnit.SECONDS);
              } catch (TimeoutException e) {
                  throw new InterruptException(0);
              }
              try {
                  synchronized (m_createLock) {
                      if (SNAP_LOG.isDebugEnabled()) {
                          SNAP_LOG.debug("Found tasks for HSIds: "
                                  + CoreUtils.hsIdCollectionToString(m_taskListsForHSIds.keySet()));
                          SNAP_LOG.debug("Looking for local HSID: " + CoreUtils.hsIdToString(context.getSiteId()));
                      }
                      Deque<SnapshotTableTask> taskList = m_taskListsForHSIds.remove(context.getSiteId());
                      // If createSetup failed, then the first site to reach here is going
                      // to send the results table generated by createSetup, and then empty out the table.
                      // All other sites to reach here will send the appropriate empty table.
                      // If createSetup was a success but the taskList is null, then we'll use the block
                      // switch to figure out what flavor of empty SnapshotSave result table to return.
                      if (!m_createSuccess.get()) {
                          // There shouldn't be any work for any site if we failed
                          assert(m_taskListsForHSIds.isEmpty());
                          VoltTable finalresult = m_createResult.get();
                          if (finalresult != null) {
                              m_createResult.set(null);
                              earlyResultTable = finalresult;
                          }
                          else {
                              // We returned a non-empty NodeResultsTable with the failures in it,
                              // every other site needs to return a NodeResultsTable as well.
                              earlyResultTable = SnapshotUtil.constructNodeResultsTable();
                          }
                      }
                      else if (taskList == null) {
                          SNAP_LOG.debug("No task for this site, block " + block);
                          // This node is participating in the snapshot but this site has nothing to do.
                          // Send back an appropriate empty table based on the block flag
                          if (block != 0) {
                              runPostTasks = true;
                              earlyResultTable = SnapshotUtil.constructPartitionResultsTable();
                              earlyResultTable.addRow(context.getHostId(), hostname,
                                      CoreUtils.getSiteIdFromHSId(context.getSiteId()), "SUCCESS", "");
                          } else {
                              //If doing snapshot for only replicated table(s), earlyResultTable here
                              //may not be empty even if the taskList of this site is null.
                              //In that case, snapshot result is preserved by earlyResultTable.
                              earlyResultTable = result;
                          }
                      }
                      else {
                          context.getSiteSnapshotConnection().initiateSnapshots(
                                  format,
                                  taskList,
                                  multiPartTxnId,
                                  m_isTruncation,
                                  m_allLocalSiteSnapshotDigestData);
                      }
  
                      if (m_deferredSetupFuture != null && taskList != null) {
                          // Add a listener to the deferred setup so that it can kick off the snapshot
                          // task once the setup is done.
                          m_deferredSetupFuture.addListener(new Runnable() {
                              @Override
                              public void run()
                              {
                                  DeferredSnapshotSetup deferredSnapshotSetup = null;
                                  try {
                                      deferredSnapshotSetup = m_deferredSetupFuture.get();
                                  } catch (Exception e) {
                                      // it doesn't throw
                                  }
  
                                  assert deferredSnapshotSetup != null;
                                  if (m_isTruncation && deferredSnapshotSetup.getError() != null) {
                                      VoltDB.crashLocalVoltDB("Unexpected exception while attempting to create truncation snapshot headers",
                                              true, deferredSnapshotSetup.getError());
                                  }
                                  context.getSiteSnapshotConnection().startSnapshotWithTargets(
                                          deferredSnapshotSetup.getPlan().getSnapshotDataTargets());
                              }
                          }, CoreUtils.SAMETHREADEXECUTOR);
                      }
                  }
              } finally {
                  SnapshotSiteProcessor.m_snapshotCreateFinishBarrier.await(120, TimeUnit.SECONDS);
              }
          } catch (TimeoutException e) {
              VoltDB.crashLocalVoltDB(
                      "Timed out waiting 120 seconds for all threads to arrive and start snapshot", true, null);
          } catch (InterruptedException | InterruptException | BrokenBarrierException | IllegalArgumentException e) {
              result.addRow(
                      context.getHostId(),
                      hostname,
                      "",
                      "FAILURE",
                      CoreUtils.throwableToString(e));
              earlyResultTable = result;
          }
  
          // If earlyResultTable is set, return here
          if (earlyResultTable != null) {
              if (runPostTasks) {
                  // Need to run post-snapshot tasks before finishing
                  SnapshotSiteProcessor.runPostSnapshotTasks(context);
              }
              return earlyResultTable;
          }
  
          if (block != 0) {
              HashSet<Exception> failures = Sets.newHashSet();
              String status = "SUCCESS";
              String err = "";
              try {
                  // For blocking snapshot, propogate the error from deferred setup back to the client
                  final DeferredSnapshotSetup deferredSnapshotSetup = m_deferredSetupFuture.get();
                  if (deferredSnapshotSetup != null && deferredSnapshotSetup.getError() != null) {
                      status = "FAILURE";
                      err = deferredSnapshotSetup.getError().toString();
                      failures.add(deferredSnapshotSetup.getError());
                  }
  
                  failures.addAll(context.getSiteSnapshotConnection().completeSnapshotWork());
                  SnapshotSiteProcessor.runPostSnapshotTasks(context);
              } catch (Exception e) {
                  status = "FAILURE";
                  err = e.toString();
                  failures.add(e);
              }
              final VoltTable blockingResult = SnapshotUtil.constructPartitionResultsTable();
  
              if (failures.isEmpty()) {
                  blockingResult.addRow(
                          context.getHostId(),
                          hostname,
                          CoreUtils.getSiteIdFromHSId(context.getSiteId()),
                          status,
                          err);
              } else {
                  status = "FAILURE";
                  for (Exception e : failures) {
                      err = e.toString();
                  }
                  blockingResult.addRow(
                          context.getHostId(),
                          hostname,
                          CoreUtils.getSiteIdFromHSId(context.getSiteId()),
                          status,
                          err);
              }
              return blockingResult;
          }
  
          return result;
      }