gt:                         VoltTable finalresult = m_createResult.get();
context: /* This file is part of VoltDB.
   * Copyright (C) 2008-2016 VoltDB Inc.
   *
   * This program is free software: you can redistribute it and/or modify
   * it under the terms of the GNU Affero General Public License as
   * published by the Free Software Foundation, either version 3 of the
   * License, or (at your option) any later version.
   *
   * This program is distributed in the hope that it will be useful,
   * but WITHOUT ANY WARRANTY; without even the implied warranty of
   * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   * GNU Affero General Public License for more details.
   *
   * You should have received a copy of the GNU Affero General Public License
   * along with VoltDB.  If not, see <http://www.gnu.org/licenses/>.
   */
  
  package org.voltdb;
  
  import java.util.ArrayDeque;
  import java.util.Deque;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Map;
  import java.util.concurrent.BrokenBarrierException;
  import java.util.concurrent.Callable;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.TimeoutException;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.concurrent.atomic.AtomicReference;
  
  import org.apache.zookeeper_voltpatches.CreateMode;
  import org.apache.zookeeper_voltpatches.KeeperException;
  import org.apache.zookeeper_voltpatches.ZooDefs.Ids;
  import org.apache.zookeeper_voltpatches.ZooKeeper;
  import org.apache.zookeeper_voltpatches.data.Stat;
  import org.json_voltpatches.JSONException;
  import org.json_voltpatches.JSONObject;
  import org.json_voltpatches.JSONStringer;
  import org.voltcore.logging.VoltLogger;
  import org.voltcore.utils.CoreUtils;
  import org.voltcore.zk.ZKUtil;
  import org.voltdb.dtxn.SiteTracker;
  import org.voltdb.iv2.TxnEgo;
  import org.voltdb.sysprocs.saverestore.CSVSnapshotWritePlan;
  import org.voltdb.sysprocs.saverestore.HashinatorSnapshotData;
  import org.voltdb.sysprocs.saverestore.IndexSnapshotWritePlan;
  import org.voltdb.sysprocs.saverestore.NativeSnapshotWritePlan;
  import org.voltdb.sysprocs.saverestore.SnapshotUtil;
  import org.voltdb.sysprocs.saverestore.SnapshotWritePlan;
  import org.voltdb.sysprocs.saverestore.StreamSnapshotWritePlan;
  
  import com.google_voltpatches.common.base.Charsets;
  import com.google_voltpatches.common.collect.Sets;
  import com.google_voltpatches.common.util.concurrent.ListenableFuture;
  import org.voltdb.sysprocs.saverestore.SnapshotPathType;
  
  /**
   * SnapshotSaveAPI extracts reusuable snapshot production code
   * that can be called from the SnapshotSave stored procedure or
   * directly from an ExecutionSite thread, perhaps has a message
   * or failure action.
   */
  public class SnapshotSaveAPI
  {
      private static final VoltLogger TRACE_LOG = new VoltLogger(SnapshotSaveAPI.class.getName());
      private static final VoltLogger SNAP_LOG = new VoltLogger("SNAPSHOT");
  
      // ugh, ick, ugh
      // IV2 does not do any snapshot work on any sites on a rejoining node.
      // The new snapshot planning stuff post-3.0 depends on it.
      public static final AtomicInteger recoveringSiteCount = new AtomicInteger(0);
  
      /**
       * Global collections populated by snapshot creator, poll'd by individual sites
       */
      // The four items containing createSetup artifacts below are all synchronized on m_createLock.
      private static final Object m_createLock = new Object();
      private static final Map<Long, Deque<SnapshotTableTask>> m_taskListsForHSIds =
          new HashMap<Long, Deque<SnapshotTableTask>>();
      private static final AtomicReference<VoltTable> m_createResult = new AtomicReference<VoltTable>();
      private static final AtomicBoolean m_createSuccess = new AtomicBoolean(false);
      private static ListenableFuture<DeferredSnapshotSetup> m_deferredSetupFuture = null;
  
      //Protected by SnapshotSiteProcessor.m_snapshotCreateLock when accessed from SnapshotSaveAPI.startSnanpshotting
      private static Map<Integer, Long> m_partitionLastSeenTransactionIds =
              new HashMap<Integer, Long>();
      private static Map<Integer, JSONObject> m_remoteDataCenterLastIds =
              new HashMap<Integer, JSONObject>();
  
      private static ExtensibleSnapshotDigestData m_allLocalSiteSnapshotDigestData;
      /**
       * The only public method: do all the work to start a snapshot.
       * Assumes that a snapshot is feasible, that the caller has validated it can
       * be accomplished, that the caller knows this is a consistent or useful
       * transaction point at which to snapshot.
       *
       * @param file_path
       * @param file_nonce
       * @param format
       * @param block
       * @param txnId
       * @param data
       * @param context
       * @param hostname
       * @return VoltTable describing the results of the snapshot attempt
       */
      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)
      {
          TRACE_LOG.trace("Creating snapshot target and handing to EEs");
          final VoltTable result = SnapshotUtil.constructNodeResultsTable();
          final int numLocalSites = context.getCluster().getDeployment().get("deployment").getSitesperhost();
          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);
                          }
                      }
  
                      m_allLocalSiteSnapshotDigestData = new ExtensibleSnapshotDigestData(
                              SnapshotSiteProcessor.getExportSequenceNumbers(),
                              SnapshotSiteProcessor.getDRTupleStreamStateInfo(),
                              remoteDataCenterLastIds);
                      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
              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.populateSequenceNumbersForExecutionSite(context);
              Integer partitionId = TxnEgo.getPartitionId(partitionTxnId);
              SNAP_LOG.debug("Registering transaction id " + partitionTxnId + " for " + TxnEgo.getPartitionId(partitionTxnId));
              m_partitionLastSeenTransactionIds.put(partitionId, partitionTxnId);
              m_remoteDataCenterLastIds.put(partitionId, perSiteRemoteDataCenterDrIds);
          }
  
          boolean runPostTasks = false;
          VoltTable earlyResultTable = null;
          try {
              SnapshotSiteProcessor.m_snapshotCreateSetupBarrier.await();
              try {
                  synchronized (m_createLock) {
                      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());
  