/*

  * Licensed to the Apache Software Foundation (ASF) under one

  * or more contributor license agreements.  See the NOTICE file

  * distributed with this work for additional information

  * regarding copyright ownership.  The ASF licenses this file

  * to you under the Apache License, Version 2.0 (the

  * "License"); you may not use this file except in compliance

  * with the License.  You may obtain a copy of the License at

  *

  *     http://www.apache.org/licenses/LICENSE-2.0

  *

  * Unless required by applicable law or agreed to in writing, software

  * distributed under the License is distributed on an "AS IS" BASIS,

  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

  * See the License for the specific language governing permissions and

  * limitations under the License.

  */

 package org.apache.giraph.master;

 import com.google.common.collect.Lists;

 import com.google.common.collect.Sets;

 import net.iharder.Base64;

 import org.apache.commons.io.FilenameUtils;

 import org.apache.giraph.bsp.ApplicationState;

 import org.apache.giraph.bsp.BspInputFormat;

 import org.apache.giraph.bsp.BspService;

 import org.apache.giraph.bsp.CentralizedServiceMaster;

 import org.apache.giraph.bsp.SuperstepState;

 import org.apache.giraph.bsp.checkpoints.CheckpointStatus;

 import org.apache.giraph.bsp.checkpoints.CheckpointSupportedChecker;

 import org.apache.giraph.comm.MasterClient;

 import org.apache.giraph.comm.MasterServer;

 import org.apache.giraph.comm.netty.NettyClient;

 import org.apache.giraph.comm.netty.NettyMasterClient;

 import org.apache.giraph.comm.netty.NettyMasterServer;

 import org.apache.giraph.comm.requests.AddressesAndPartitionsRequest;

 import org.apache.giraph.conf.GiraphConfiguration;

 import org.apache.giraph.conf.GiraphConstants;

 import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;

 import org.apache.giraph.counters.CustomCounter;

 import org.apache.giraph.counters.CustomCounters;

 import org.apache.giraph.counters.GiraphCountersThriftStruct;

 import org.apache.giraph.counters.GiraphStats;

 import org.apache.giraph.counters.GiraphTimers;

 import org.apache.giraph.graph.AddressesAndPartitionsWritable;

 import org.apache.giraph.graph.GlobalStats;

 import org.apache.giraph.graph.GraphFunctions;

 import org.apache.giraph.graph.GraphState;

 import org.apache.giraph.graph.GraphTaskManager;

 import org.apache.giraph.io.EdgeInputFormat;

 import org.apache.giraph.io.GiraphInputFormat;

 import org.apache.giraph.io.InputType;

 import org.apache.giraph.io.MappingInputFormat;

 import org.apache.giraph.io.VertexInputFormat;

 import org.apache.giraph.master.input.MasterInputSplitsHandler;

 import org.apache.giraph.metrics.AggregatedMetrics;

 import org.apache.giraph.metrics.GiraphMetrics;

 import org.apache.giraph.metrics.GiraphTimer;

 import org.apache.giraph.metrics.GiraphTimerContext;

 import org.apache.giraph.metrics.ResetSuperstepMetricsObserver;

 import org.apache.giraph.metrics.SuperstepMetricsRegistry;

 import org.apache.giraph.metrics.WorkerSuperstepMetrics;

 import org.apache.giraph.partition.BasicPartitionOwner;

 import org.apache.giraph.partition.MasterGraphPartitioner;

 import org.apache.giraph.partition.PartitionOwner;

 import org.apache.giraph.partition.PartitionStats;

 import org.apache.giraph.partition.PartitionUtils;

 import org.apache.giraph.time.SystemTime;

 import org.apache.giraph.time.Time;

 import org.apache.giraph.utils.CheckpointingUtils;

 import org.apache.giraph.utils.JMapHistoDumper;

 import org.apache.giraph.utils.ReactiveJMapHistoDumper;

 import org.apache.giraph.utils.ReflectionUtils;

 import org.apache.giraph.utils.WritableUtils;

 import org.apache.giraph.worker.WorkerInfo;

 import org.apache.giraph.zk.BspEvent;

 import org.apache.giraph.zk.PredicateLock;

 import org.apache.hadoop.fs.FSDataOutputStream;

 import org.apache.hadoop.fs.FileSystem;

 import org.apache.hadoop.fs.Path;

 import org.apache.hadoop.io.Writable;

 import org.apache.hadoop.io.WritableComparable;

 import org.apache.hadoop.mapred.JobID;

 import org.apache.hadoop.mapred.RunningJob;

 import org.apache.hadoop.mapreduce.Counter;

 import org.apache.hadoop.mapreduce.InputSplit;

 import org.apache.hadoop.mapreduce.Mapper;

 import org.apache.log4j.Logger;

 import org.apache.zookeeper.CreateMode;

 import org.apache.zookeeper.KeeperException;

 import org.apache.zookeeper.WatchedEvent;

 import org.apache.zookeeper.Watcher.Event.EventType;

 import org.apache.zookeeper.ZooDefs.Ids;

 import org.json.JSONArray;

 import org.json.JSONException;

 import org.json.JSONObject;

 import java.io.DataInputStream;

 import java.io.IOException;

 import java.io.PrintStream;

 import java.nio.charset.Charset;

 import java.util.ArrayList;

 import java.util.Collection;

 import java.util.Collections;

 import java.util.Comparator;

 import java.util.HashSet;

 import java.util.List;

 import java.util.Map;

 import java.util.Set;

 import java.util.TreeSet;

 import java.util.concurrent.TimeUnit;

 import static org.apache.giraph.conf.GiraphConstants.INPUT_SPLIT_SAMPLE_PERCENT;

 import static org.apache.giraph.conf.GiraphConstants.KEEP_ZOOKEEPER_DATA;

 import static org.apache.giraph.conf.GiraphConstants.PARTITION_LONG_TAIL_MIN_PRINT;

 /**

  * ZooKeeper-based implementation of {@link CentralizedServiceMaster}.

  *

  * @param <I> Vertex id

  * @param <V> Vertex data

  * @param <E> Edge data

  */

 @SuppressWarnings("rawtypes, unchecked")

 public class BspServiceMaster<I extends WritableComparable,

     V extends Writable, E extends Writable>

     extends BspService<I, V, E>

     implements CentralizedServiceMaster<I, V, E>,

     ResetSuperstepMetricsObserver {

   /** Print worker names only if there are 10 workers left */

   public static final int MAX_PRINTABLE_REMAINING_WORKERS = 10;

   /** How many threads to use when writing input splits to zookeeper*/

   public static final String NUM_MASTER_ZK_INPUT_SPLIT_THREADS =

       "giraph.numMasterZkInputSplitThreads";

   /** Default number of threads to use when writing input splits to zookeeper */

   public static final int DEFAULT_INPUT_SPLIT_THREAD_COUNT = 1;

   /** Time instance to use for timing */

   private static final Time TIME = SystemTime.get();

   /** Class logger */

   private static final Logger LOG = Logger.getLogger(BspServiceMaster.class);

   /** Am I the master? */

   private boolean isMaster = false;

   /** Max number of workers */

   private final int maxWorkers;

   /** Min number of workers */

   private final int minWorkers;

   /** Max number of supersteps */

   private final int maxNumberOfSupersteps;

   /** Min % responded workers */

   private final float minPercentResponded;

   /** Msecs to wait for an event */

   private final int eventWaitMsecs;

   /** Max msecs to wait for a superstep to get enough workers */

   private final int maxSuperstepWaitMsecs;

   /** Max msecs to wait for the workers to write their counters for a

    * superstep*/

   private final int maxCounterWaitMsecs;

   /** Min number of long tails before printing */

   private final int partitionLongTailMinPrint;

   /** Last finalized checkpoint */

   private long lastCheckpointedSuperstep = -1;

   /** Worker wrote checkpoint */

   private final BspEvent workerWroteCheckpoint;

   /** State of the superstep changed */

   private final BspEvent superstepStateChanged;

   /** Master graph partitioner */

   private final MasterGraphPartitioner<I, V, E> masterGraphPartitioner;

   /** All the partition stats from the last superstep */

   private final List<PartitionStats> allPartitionStatsList =

       new ArrayList<PartitionStats>();

   /** Handler for global communication */

   private MasterGlobalCommHandler globalCommHandler;

   /** Handler for aggregators to reduce/broadcast translation */

   private AggregatorToGlobalCommTranslation aggregatorTranslation;

   /** Master class */

   private MasterCompute masterCompute;

   /** IPC Client */

   private MasterClient masterClient;

   /** IPC Server */

   private MasterServer masterServer;

   /** Master info */

   private MasterInfo masterInfo;

   /** List of workers in current superstep, sorted by task id */

   private List<WorkerInfo> chosenWorkerInfoList = Lists.newArrayList();

   /** Observers over master lifecycle. */

   private final MasterObserver[] observers;

   // Per-Superstep Metrics

   /** MasterCompute time */

   private GiraphTimer masterComputeTimer;

   /** Checkpoint frequency */

   private final int checkpointFrequency;

   /** Current checkpoint status */

   private CheckpointStatus checkpointStatus;

   /** Checks if checkpointing supported */

   private final CheckpointSupportedChecker checkpointSupportedChecker;

   /** Thrift struct to store the aggregate counters */

   private final GiraphCountersThriftStruct giraphCountersThriftStruct;

   /**

    * Constructor for setting up the master.

    *

    * @param context Mapper context

    * @param graphTaskManager GraphTaskManager for this compute node

    */

   public BspServiceMaster(

       Mapper<?, ?, ?, ?>.Context context,

       GraphTaskManager<I, V, E> graphTaskManager) {

     super(context, graphTaskManager);

     workerWroteCheckpoint = new PredicateLock(context);

     registerBspEvent(workerWroteCheckpoint);

     superstepStateChanged = new PredicateLock(context);

     registerBspEvent(superstepStateChanged);

     ImmutableClassesGiraphConfiguration<I, V, E> conf =

         getConfiguration();

     maxWorkers = conf.getMaxWorkers();

     minWorkers = conf.getMinWorkers();

     maxNumberOfSupersteps = conf.getMaxNumberOfSupersteps();

     minPercentResponded = GiraphConstants.MIN_PERCENT_RESPONDED.get(conf);

     eventWaitMsecs = conf.getEventWaitMsecs();

     maxSuperstepWaitMsecs = conf.getMaxMasterSuperstepWaitMsecs();

     maxCounterWaitMsecs = conf.getMaxCounterWaitMsecs();

     partitionLongTailMinPrint = PARTITION_LONG_TAIL_MIN_PRINT.get(conf);

     masterGraphPartitioner =

         getGraphPartitionerFactory().createMasterGraphPartitioner();

     if (conf.isJMapHistogramDumpEnabled()) {

       conf.addMasterObserverClass(JMapHistoDumper.class);

     }

     if (conf.isReactiveJmapHistogramDumpEnabled()) {

       conf.addMasterObserverClass(ReactiveJMapHistoDumper.class);

     }

     observers = conf.createMasterObservers(context);

     this.checkpointFrequency = conf.getCheckpointFrequency();

     this.checkpointStatus = CheckpointStatus.NONE;

     this.checkpointSupportedChecker =

         ReflectionUtils.newInstance(

             GiraphConstants.CHECKPOINT_SUPPORTED_CHECKER.get(conf));

     this.giraphCountersThriftStruct = new GiraphCountersThriftStruct();

     GiraphMetrics.get().addSuperstepResetObserver(this);

     GiraphStats.init(context);

   }

   @Override

   public void newSuperstep(SuperstepMetricsRegistry superstepMetrics) {

     masterComputeTimer = new GiraphTimer(superstepMetrics,

         "master-compute-call", TimeUnit.MILLISECONDS);

   }

   @Override

   public void setJobState(ApplicationState state,

       long applicationAttempt,

       long desiredSuperstep) {

     setJobState(state, applicationAttempt, desiredSuperstep, true);

   }

   /**

    * Set the job state.

    *

    * @param state State of the application.

    * @param applicationAttempt Attempt to start on

    * @param desiredSuperstep Superstep to restart from (if applicable)

    * @param killJobOnFailure if true, and the desired state is FAILED,

    *                         then kill this job.

    */

   private void setJobState(ApplicationState state,

       long applicationAttempt,

       long desiredSuperstep,

       boolean killJobOnFailure) {

     JSONObject jobState = new JSONObject();

     try {

       jobState.put(JSONOBJ_STATE_KEY, state.toString());

       jobState.put(JSONOBJ_APPLICATION_ATTEMPT_KEY, applicationAttempt);

       jobState.put(JSONOBJ_SUPERSTEP_KEY, desiredSuperstep);

     } catch (JSONException e) {

       throw new RuntimeException("setJobState: Couldn't put " +

           state.toString());

     }

     if (LOG.isInfoEnabled()) {

       LOG.info("setJobState: " + jobState.toString() + " on superstep " +

           getSuperstep());

     }

     try {

       getZkExt().createExt(masterJobStatePath + "/jobState",

           jobState.toString().getBytes(Charset.defaultCharset()),

           Ids.OPEN_ACL_UNSAFE,

           CreateMode.PERSISTENT_SEQUENTIAL,

           true);

       LOG.info("setJobState: " + jobState);

     } catch (KeeperException.NodeExistsException e) {

       throw new IllegalStateException(

           "setJobState: Imposible that " +

               masterJobStatePath + " already exists!", e);

     } catch (KeeperException e) {

       throw new IllegalStateException(

           "setJobState: Unknown KeeperException for " +

               masterJobStatePath, e);

     } catch (InterruptedException e) {

       throw new IllegalStateException(

           "setJobState: Unknown InterruptedException for " +

               masterJobStatePath, e);

     }

     if (state == ApplicationState.FAILED && killJobOnFailure) {

       failJob(new IllegalStateException("FAILED"));

     }

   }

   /**

    * Set the job state to FAILED. This will kill the job, and log exceptions to

    * any observers.

    *

    * @param reason The reason the job failed

    */

   private void setJobStateFailed(String reason) {

     getGraphTaskManager().getJobProgressTracker().logFailure(reason);

     setJobState(ApplicationState.FAILED, -1, -1, false);

     failJob(new IllegalStateException(reason));

   }

   /**

    * Common method for generating vertex/edge input splits.

    *

    * @param inputFormat The vertex/edge input format

    * @param minSplitCountHint Minimum number of splits to create (hint)

    * @param inputSplitType Type of input splits (for logging purposes)

    * @return List of input splits for the given format

    */

   private List<InputSplit> generateInputSplits(GiraphInputFormat inputFormat,

                                                int minSplitCountHint,

                                                InputType inputSplitType) {

     String logPrefix = "generate" + inputSplitType + "InputSplits";

     List<InputSplit> splits;

     try {

       splits = inputFormat.getSplits(getContext(), minSplitCountHint);

     } catch (IOException e) {

       throw new IllegalStateException(logPrefix + ": Got IOException", e);

     } catch (InterruptedException e) {

       throw new IllegalStateException(

           logPrefix + ": Got InterruptedException", e);

     }

     float samplePercent =

         INPUT_SPLIT_SAMPLE_PERCENT.get(getConfiguration());

     if (samplePercent != INPUT_SPLIT_SAMPLE_PERCENT.getDefaultValue()) {

       int lastIndex = (int) (samplePercent * splits.size() / 100f);

       Collections.shuffle(splits);

       List<InputSplit> sampleSplits = splits.subList(0, lastIndex);

       LOG.warn(logPrefix + ": Using sampling - Processing only " +

           sampleSplits.size() + " instead of " + splits.size() +

           " expected splits.");

       return sampleSplits;

     } else {

       if (LOG.isInfoEnabled()) {

         LOG.info(logPrefix + ": Got " + splits.size() +

             " input splits for " + minSplitCountHint + " input threads");

       }

       return splits;

     }

   }

   /**

    * When there is no salvaging this job, fail it.

    *

    * @param e Exception to log to observers

    */

   private void failJob(Exception e) {

     LOG.fatal("failJob: Killing job " + getJobId());

     LOG.fatal("failJob: exception " + e.toString());

     try {

       if (getConfiguration().isPureYarnJob()) {

         throw new RuntimeException(

           "BspServiceMaster (YARN profile) is " +

           "FAILING this task, throwing exception to end job run.", e);

       } else {

         @SuppressWarnings("deprecation")

         org.apache.hadoop.mapred.JobClient jobClient =

           new org.apache.hadoop.mapred.JobClient(

             (org.apache.hadoop.mapred.JobConf)

             getContext().getConfiguration());

         try {

           @SuppressWarnings("deprecation")

           JobID jobId = JobID.forName(getJobId());

           RunningJob job = jobClient.getJob(jobId);

           if (job != null) {

             job.killJob();

           } else {

             LOG.error("Job not found for jobId=" + getJobId());

           }

         } catch (IllegalArgumentException iae) {

           LOG.info("This job (" + getJobId() +

                        ") is not a legacy Hadoop job and will " +

                        "continue with failure cleanup." +

                        e.getMessage(),

                    e);

         }

       }

     } catch (IOException ioe) {

       throw new RuntimeException(ioe);

     } finally {

       failureCleanup(e);

     }

   }

   /**

    * Parse the {@link WorkerInfo} objects from a ZooKeeper path

    * (and children).

    *

    * @param workerInfosPath Path where all the workers are children

    * @param watch Watch or not?

    * @return List of workers in that path

    */

   private List<WorkerInfo> getWorkerInfosFromPath(String workerInfosPath,

       boolean watch) {

     List<WorkerInfo> workerInfoList = new ArrayList<WorkerInfo>();

     List<String> workerInfoPathList;

     try {

       workerInfoPathList =

           getZkExt().getChildrenExt(workerInfosPath, watch, false, true);

     } catch (KeeperException e) {

       throw new IllegalStateException(

           "getWorkers: Got KeeperException", e);

     } catch (InterruptedException e) {

       throw new IllegalStateException(

           "getWorkers: Got InterruptedStateException", e);

     }

     for (String workerInfoPath : workerInfoPathList) {

       WorkerInfo workerInfo = new WorkerInfo();

       try {

         WritableUtils.readFieldsFromZnode(

             getZkExt(), workerInfoPath, true, null, workerInfo);

         workerInfoList.add(workerInfo);

       } catch (IllegalStateException e) {

         LOG.warn("Can't get info from worker, did it die in between? " +

             "workerInfoPath=" + workerInfoPath, e);

       }

     }

     return workerInfoList;

   }

   /**

    * Get the healthy and unhealthy {@link WorkerInfo} objects for

    * a superstep

    *

    * @param superstep superstep to check

    * @param healthyWorkerInfoList filled in with current data

    * @param unhealthyWorkerInfoList filled in with current data

    */

   private void getAllWorkerInfos(

       long superstep,

       List<WorkerInfo> healthyWorkerInfoList,

       List<WorkerInfo> unhealthyWorkerInfoList) {

     String healthyWorkerInfoPath =

         getWorkerInfoHealthyPath(getApplicationAttempt(), superstep);

     String unhealthyWorkerInfoPath =

         getWorkerInfoUnhealthyPath(getApplicationAttempt(), superstep);

     try {

       getZkExt().createOnceExt(healthyWorkerInfoPath,

           null,

           Ids.OPEN_ACL_UNSAFE,

           CreateMode.PERSISTENT,

           true);

     } catch (KeeperException e) {

       throw new IllegalStateException("getWorkers: KeeperException", e);

     } catch (InterruptedException e) {

       throw new IllegalStateException("getWorkers: InterruptedException", e);

     }

     try {

       getZkExt().createOnceExt(unhealthyWorkerInfoPath,

           null,

           Ids.OPEN_ACL_UNSAFE,

           CreateMode.PERSISTENT,

           true);

     } catch (KeeperException e) {

       throw new IllegalStateException("getWorkers: KeeperException", e);

     } catch (InterruptedException e) {

       throw new IllegalStateException("getWorkers: InterruptedException", e);

     }

     List<WorkerInfo> currentHealthyWorkerInfoList =

         getWorkerInfosFromPath(healthyWorkerInfoPath, true);

     List<WorkerInfo> currentUnhealthyWorkerInfoList =

         getWorkerInfosFromPath(unhealthyWorkerInfoPath, false);

     healthyWorkerInfoList.clear();

     if (currentHealthyWorkerInfoList != null) {

       for (WorkerInfo healthyWorkerInfo :

         currentHealthyWorkerInfoList) {

         healthyWorkerInfoList.add(healthyWorkerInfo);

       }

     }

     unhealthyWorkerInfoList.clear();

     if (currentUnhealthyWorkerInfoList != null) {

       for (WorkerInfo unhealthyWorkerInfo :

         currentUnhealthyWorkerInfoList) {

         unhealthyWorkerInfoList.add(unhealthyWorkerInfo);

       }

     }

   }

   @Override

   public List<WorkerInfo> checkWorkers() {

     boolean failJob = true;

     long failWorkerCheckMsecs =

         SystemTime.get().getMilliseconds() + maxSuperstepWaitMsecs;

     List<WorkerInfo> healthyWorkerInfoList = new ArrayList<WorkerInfo>();

     List<WorkerInfo> unhealthyWorkerInfoList = new ArrayList<WorkerInfo>();

     int totalResponses = -1;

     while (SystemTime.get().getMilliseconds() < failWorkerCheckMsecs) {

       getContext().progress();

       getAllWorkerInfos(

           getSuperstep(), healthyWorkerInfoList, unhealthyWorkerInfoList);

       totalResponses = healthyWorkerInfoList.size() +

           unhealthyWorkerInfoList.size();

       if ((totalResponses * 100.0f / maxWorkers) >=

           minPercentResponded) {

         failJob = false;

         break;

       }

       getContext().setStatus(getGraphTaskManager().getGraphFunctions() + " " +

           "checkWorkers: Only found " +

           totalResponses +

           " responses of " + maxWorkers +

           " needed to start superstep " +

           getSuperstep());

       if (getWorkerHealthRegistrationChangedEvent().waitMsecs(

           eventWaitMsecs)) {

         if (LOG.isDebugEnabled()) {

           LOG.debug("checkWorkers: Got event that health " +

               "registration changed, not using poll attempt");

         }

         getWorkerHealthRegistrationChangedEvent().reset();

         continue;

       }

       if (LOG.isInfoEnabled()) {

         LOG.info("checkWorkers: Only found " + totalResponses +

             " responses of " + maxWorkers +

             " needed to start superstep " +

             getSuperstep() + ".  Reporting every " +

             eventWaitMsecs + " msecs, " +

             (failWorkerCheckMsecs - SystemTime.get().getMilliseconds()) +

             " more msecs left before giving up.");

         // Find the missing workers if there are only a few

         if ((maxWorkers - totalResponses) <=

             partitionLongTailMinPrint) {

           logMissingWorkersOnSuperstep(healthyWorkerInfoList,

               unhealthyWorkerInfoList);

         }

       }

     }

     if (failJob) {

       LOG.error("checkWorkers: Did not receive enough processes in " +

           "time (only " + totalResponses + " of " +

           minWorkers + " required) after waiting " + maxSuperstepWaitMsecs +

           "msecs).  This occurs if you do not have enough map tasks " +

           "available simultaneously on your Hadoop instance to fulfill " +

           "the number of requested workers.");

       return null;

     }

     if (healthyWorkerInfoList.size() < minWorkers) {

       LOG.error("checkWorkers: Only " + healthyWorkerInfoList.size() +

           " available when " + minWorkers + " are required.");

       logMissingWorkersOnSuperstep(healthyWorkerInfoList,

           unhealthyWorkerInfoList);

       return null;

     }

     getContext().setStatus(getGraphTaskManager().getGraphFunctions() + " " +

         "checkWorkers: Done - Found " + totalResponses +

         " responses of " + maxWorkers + " needed to start superstep " +

         getSuperstep());

     return healthyWorkerInfoList;

   }

   /**

    * Log info level of the missing workers on the superstep

    *

    * @param healthyWorkerInfoList Healthy worker list

    * @param unhealthyWorkerInfoList Unhealthy worker list

    */

   private void logMissingWorkersOnSuperstep(

       List<WorkerInfo> healthyWorkerInfoList,

       List<WorkerInfo> unhealthyWorkerInfoList) {

     if (LOG.isInfoEnabled()) {

       Set<Integer> partitionSet = new TreeSet<Integer>();

       for (WorkerInfo workerInfo : healthyWorkerInfoList) {

         partitionSet.add(workerInfo.getTaskId() % maxWorkers);

       }

       for (WorkerInfo workerInfo : unhealthyWorkerInfoList) {

         partitionSet.add(workerInfo.getTaskId() % maxWorkers);

       }

       for (int i = 1; i <= maxWorkers; ++i) {

         if (partitionSet.contains(Integer.valueOf(i))) {

           continue;

         } else if (i == getTaskId() % maxWorkers) {

           continue;

         } else {

           LOG.info("logMissingWorkersOnSuperstep: No response from " +

               "partition " + i + " (could be master)");

         }

       }

     }

   }

   /**

    * Common method for creating vertex/edge input splits.

    *

    * @param inputFormat The vertex/edge input format

    * @param inputSplitType Type of input split (for logging purposes)

    * @return Number of splits. Returns -1 on failure to create

    *         valid input splits.

    */

   private int createInputSplits(GiraphInputFormat inputFormat,

                                 InputType inputSplitType) {

     ImmutableClassesGiraphConfiguration conf = getConfiguration();

     String logPrefix = "create" + inputSplitType + "InputSplits";

     // Only the 'master' should be doing this.  Wait until the number of

     // processes that have reported health exceeds the minimum percentage.

     // If the minimum percentage is not met, fail the job.  Otherwise

     // generate the input splits

     List<WorkerInfo> healthyWorkerInfoList = checkWorkers();

     if (healthyWorkerInfoList == null) {

       setJobStateFailed("Not enough healthy workers to create input splits");

       return -1;

     }

     globalCommHandler.getInputSplitsHandler().initialize(masterClient,

         healthyWorkerInfoList);

     // Create at least as many splits as the total number of input threads.

     int minSplitCountHint = healthyWorkerInfoList.size() *

         conf.getNumInputSplitsThreads();

     // Note that the input splits may only be a sample if

     // INPUT_SPLIT_SAMPLE_PERCENT is set to something other than 100

     List<InputSplit> splitList = generateInputSplits(inputFormat,

         minSplitCountHint, inputSplitType);

     if (splitList.isEmpty() && GiraphConstants.FAIL_ON_EMPTY_INPUT.get(conf)) {

       LOG.fatal(logPrefix + ": Failing job due to 0 input splits, " +

           "check input of " + inputFormat.getClass().getName() + "!");

       getContext().setStatus("Failing job due to 0 input splits, " +

           "check input of " + inputFormat.getClass().getName() + "!");

       setJobStateFailed("******* PLEASE CHECK YOUR INPUT TABLES - PARTITIONS " +

           "WHICH YOU SPECIFIED ARE MISSING (for " + inputSplitType +

           " input). FAILING THE JOB *******");

     }

     if (minSplitCountHint > splitList.size()) {

       LOG.warn(logPrefix + ": Number of inputSplits=" +

           splitList.size() + " < " +

           minSplitCountHint +

           "=total number of input threads, " +

           "some threads will be not used");

     }

     globalCommHandler.getInputSplitsHandler().addSplits(inputSplitType,

         splitList, inputFormat);

     return splitList.size();

   }

   @Override

   public int createMappingInputSplits() {

     if (!getConfiguration().hasMappingInputFormat()) {

       return 0;

     }

     MappingInputFormat<I, V, E, ? extends Writable> mappingInputFormat =

       getConfiguration().createWrappedMappingInputFormat();

     return createInputSplits(mappingInputFormat, InputType.MAPPING);

   }

   @Override

   public int createVertexInputSplits() {

     int splits = 0;

     if (getConfiguration().hasVertexInputFormat()) {

       VertexInputFormat<I, V, E> vertexInputFormat =

           getConfiguration().createWrappedVertexInputFormat();

       splits = createInputSplits(vertexInputFormat, InputType.VERTEX);

     }

     MasterProgress.get().setVertexInputSplitCount(splits);

     getJobProgressTracker().updateMasterProgress(MasterProgress.get());

     return splits;

   }

   @Override

   public int createEdgeInputSplits() {

     int splits = 0;

     if (getConfiguration().hasEdgeInputFormat()) {

       EdgeInputFormat<I, E> edgeInputFormat =

           getConfiguration().createWrappedEdgeInputFormat();

       splits = createInputSplits(edgeInputFormat, InputType.EDGE);

     }

     MasterProgress.get().setEdgeInputSplitsCount(splits);

     getJobProgressTracker().updateMasterProgress(MasterProgress.get());

     return splits;

   }

   @Override

   public List<WorkerInfo> getWorkerInfoList() {

     return chosenWorkerInfoList;

   }

   @Override

   public MasterGlobalCommHandler getGlobalCommHandler() {

     return globalCommHandler;

   }

   @Override

   public AggregatorToGlobalCommTranslation getAggregatorTranslationHandler() {

     return aggregatorTranslation;

   }

   @Override

   public MasterCompute getMasterCompute() {

     return masterCompute;

   }

   /**

    * Read the finalized checkpoint file and associated metadata files for the

    * checkpoint.  Modifies the {@link PartitionOwner} objects to get the

    * checkpoint prefixes.  It is an optimization to prevent all workers from

    * searching all the files.  Also read in the aggregator data from the

    * finalized checkpoint file and setting it.

    *

    * @param superstep Checkpoint set to examine.

    * @throws IOException

    * @throws InterruptedException

    * @throws KeeperException

    * @return Collection of generated partition owners.

    */

   private Collection<PartitionOwner> prepareCheckpointRestart(long superstep)

     throws IOException, KeeperException, InterruptedException {

     List<PartitionOwner> partitionOwners = new ArrayList<>();

     FileSystem fs = getFs();

     String finalizedCheckpointPath = getSavedCheckpointBasePath(superstep) +

         CheckpointingUtils.CHECKPOINT_FINALIZED_POSTFIX;

     LOG.info("Loading checkpoint from " + finalizedCheckpointPath);

     DataInputStream finalizedStream =

         fs.open(new Path(finalizedCheckpointPath));

     GlobalStats globalStats = new GlobalStats();

     globalStats.readFields(finalizedStream);

     updateCounters(globalStats);

     SuperstepClasses superstepClasses =

         SuperstepClasses.createToRead(getConfiguration());

     superstepClasses.readFields(finalizedStream);

     getConfiguration().updateSuperstepClasses(superstepClasses);

     int prefixFileCount = finalizedStream.readInt();

     String checkpointFile =

         finalizedStream.readUTF();

     for (int i = 0; i < prefixFileCount; ++i) {

       int mrTaskId = finalizedStream.readInt();

       DataInputStream metadataStream = fs.open(new Path(checkpointFile +

           "." + mrTaskId + CheckpointingUtils.CHECKPOINT_METADATA_POSTFIX));

       long partitions = metadataStream.readInt();

       WorkerInfo worker = getWorkerInfoById(mrTaskId);

       for (long p = 0; p < partitions; ++p) {

         int partitionId = metadataStream.readInt();

         PartitionOwner partitionOwner = new BasicPartitionOwner(partitionId,

             worker);

         partitionOwners.add(partitionOwner);

         LOG.info("prepareCheckpointRestart partitionId=" + partitionId +

             " assigned to " + partitionOwner);

       }

       metadataStream.close();

     }

     //Ordering appears to be important as of right now we rely on this ordering

     //in WorkerGraphPartitioner

     Collections.sort(partitionOwners, new Comparator<PartitionOwner>() {

       @Override

       public int compare(PartitionOwner p1, PartitionOwner p2) {

         return Integer.compare(p1.getPartitionId(), p2.getPartitionId());

       }

     });

     globalCommHandler.getAggregatorHandler().readFields(finalizedStream);

     aggregatorTranslation.readFields(finalizedStream);

     masterCompute.readFields(finalizedStream);

     finalizedStream.close();

     return partitionOwners;

   }

   @Override

   public void setup() {

     // Might have to manually load a checkpoint.

     // In that case, the input splits are not set, they will be faked by

     // the checkpoint files.  Each checkpoint file will be an input split

     // and the input split

     if (getRestartedSuperstep() != UNSET_SUPERSTEP) {

       GiraphStats.getInstance().getSuperstepCounter().

         setValue(getRestartedSuperstep());

     }

     for (MasterObserver observer : observers) {

       observer.preApplication();

       getContext().progress();

     }

   }

   @Override

   public boolean becomeMaster() {

     // Create my bid to become the master, then try to become the worker

     // or return false.

     String myBid = null;

     try {

       myBid =

           getZkExt().createExt(masterElectionPath +

               "/" + getHostnameTaskId(),

               null,

               Ids.OPEN_ACL_UNSAFE,

               CreateMode.EPHEMERAL_SEQUENTIAL,

               true);

     } catch (KeeperException e) {

       throw new IllegalStateException(

           "becomeMaster: KeeperException", e);

     } catch (InterruptedException e) {

       throw new IllegalStateException(

           "becomeMaster: IllegalStateException", e);

     }

     while (true) {

       JSONObject jobState = getJobState();

       try {

         if ((jobState != null) &&

             ApplicationState.valueOf(

                 jobState.getString(JSONOBJ_STATE_KEY)) ==

                 ApplicationState.FINISHED) {

           LOG.info("becomeMaster: Job is finished, " +

               "give up trying to be the master!");

           isMaster = false;

           return isMaster;

         }

       } catch (JSONException e) {

         throw new IllegalStateException(

             "becomeMaster: Couldn't get state from " + jobState, e);

       }

       try {

         List<String> masterChildArr =

             getZkExt().getChildrenExt(

                 masterElectionPath, true, true, true);

         if (LOG.isInfoEnabled()) {

           LOG.info("becomeMaster: First child is '" +

               masterChildArr.get(0) + "' and my bid is '" +

               myBid + "'");

         }

         if (masterChildArr.get(0).equals(myBid)) {

           GiraphStats.getInstance().getCurrentMasterTaskPartition().

               setValue(getTaskId());

           globalCommHandler = new MasterGlobalCommHandler(

               new MasterAggregatorHandler(getConfiguration(), getContext()),

               new MasterInputSplitsHandler(

                   getConfiguration().useInputSplitLocality(), getContext()));

           aggregatorTranslation = new AggregatorToGlobalCommTranslation(

               getConfiguration(), globalCommHandler);

           globalCommHandler.getAggregatorHandler().initialize(this);

           masterCompute = getConfiguration().createMasterCompute();

           masterCompute.setMasterService(this);

           masterInfo = new MasterInfo();

           masterServer =

               new NettyMasterServer(getConfiguration(), this, getContext(),

                   getGraphTaskManager().createUncaughtExceptionHandler());

           masterInfo.setInetSocketAddress(masterServer.getMyAddress(),

               masterServer.getLocalHostOrIp());

           masterInfo.setTaskId(getTaskId());

           masterClient =

               new NettyMasterClient(getContext(), getConfiguration(), this,

                   getGraphTaskManager().createUncaughtExceptionHandler());

           masterServer.setFlowControl(masterClient.getFlowControl());

           if (LOG.isInfoEnabled()) {

             LOG.info("becomeMaster: I am now the master!");

           }

           isMaster = true;

           return isMaster;

         }

         LOG.info("becomeMaster: Waiting to become the master...");

         getMasterElectionChildrenChangedEvent().waitForTimeoutOrFail(

             GiraphConstants.WAIT_ZOOKEEPER_TIMEOUT_MSEC.get(

                 getConfiguration()));

         getMasterElectionChildrenChangedEvent().reset();

       } catch (KeeperException e) {

         throw new IllegalStateException(

             "becomeMaster: KeeperException", e);

       } catch (InterruptedException e) {

         throw new IllegalStateException(

             "becomeMaster: IllegalStateException", e);

       }

     }

   }

   @Override

   public MasterInfo getMasterInfo() {

     return masterInfo;

   }

   /**

    * Collect and aggregate the worker statistics for a particular superstep.

    *

    * @param superstep Superstep to aggregate on

    * @return Global statistics aggregated on all worker statistics

    */

   private GlobalStats aggregateWorkerStats(long superstep) {

     ImmutableClassesGiraphConfiguration conf = getConfiguration();

     GlobalStats globalStats = new GlobalStats();

     // Get the stats from the all the worker selected nodes

     String workerFinishedPath =

         getWorkerMetricsFinishedPath(getApplicationAttempt(), superstep);

     List<String> workerFinishedPathList = null;

     try {

       workerFinishedPathList =

           getZkExt().getChildrenExt(

               workerFinishedPath, false, false, true);

     } catch (KeeperException e) {

       throw new IllegalStateException(

           "aggregateWorkerStats: KeeperException", e);

     } catch (InterruptedException e) {

       throw new IllegalStateException(

           "aggregateWorkerStats: InterruptedException", e);

     }

     AggregatedMetrics aggregatedMetrics = new AggregatedMetrics();

     for (String finishedPath : workerFinishedPathList) {

       String hostnamePartitionId = FilenameUtils.getName(finishedPath);

       JSONObject workerFinishedInfoObj = null;

       try {

         byte [] zkData =

             getZkExt().getData(finishedPath, false, null);

         workerFinishedInfoObj = new JSONObject(new String(zkData,

             Charset.defaultCharset()));

         globalStats.addMessageCount(

             workerFinishedInfoObj.getLong(

                 JSONOBJ_NUM_MESSAGES_KEY));

         globalStats.addMessageBytesCount(

           workerFinishedInfoObj.getLong(

               JSONOBJ_NUM_MESSAGE_BYTES_KEY));

         if (conf.metricsEnabled() &&

             workerFinishedInfoObj.has(JSONOBJ_METRICS_KEY)) {

           WorkerSuperstepMetrics workerMetrics = new WorkerSuperstepMetrics();

           WritableUtils.readFieldsFromByteArray(

               Base64.decode(

                   workerFinishedInfoObj.getString(

                       JSONOBJ_METRICS_KEY)),

               workerMetrics);

           globalStats.addOocLoadBytesCount(

               workerMetrics.getBytesLoadedFromDisk());

           globalStats.addOocStoreBytesCount(

               workerMetrics.getBytesStoredOnDisk());

           // Find the lowest percentage of graph in memory across all workers

           // for one superstep

           globalStats.setLowestGraphPercentageInMemory(

               Math.min(globalStats.getLowestGraphPercentageInMemory(),

                   (int) Math.round(

                       workerMetrics.getGraphPercentageInMemory())));

           aggregatedMetrics.add(workerMetrics, hostnamePartitionId);

         }

       } catch (JSONException e) {

         throw new IllegalStateException(

             "aggregateWorkerStats: JSONException", e);

       } catch (KeeperException e) {

         throw new IllegalStateException(

             "aggregateWorkerStats: KeeperException", e);

       } catch (InterruptedException e) {

         throw new IllegalStateException(

             "aggregateWorkerStats: InterruptedException", e);

       } catch (IOException e) {

         throw new IllegalStateException(

             "aggregateWorkerStats: IOException", e);

       }

     }

     allPartitionStatsList.clear();

     Iterable<PartitionStats> statsList = globalCommHandler.getAllPartitionStats(

         workerFinishedPathList.size(), getContext());

     for (PartitionStats partitionStats : statsList) {

       globalStats.addPartitionStats(partitionStats);

       allPartitionStatsList.add(partitionStats);

     }

     if (conf.metricsEnabled()) {

       if (GiraphConstants.METRICS_DIRECTORY.isDefaultValue(conf)) {

         aggregatedMetrics.print(superstep, System.err);

       } else {

         printAggregatedMetricsToHDFS(superstep, aggregatedMetrics);

       }

       for (MasterObserver observer : observers) {

         observer.superstepMetricsUpdate(

             superstep, aggregatedMetrics, allPartitionStatsList);

       }

     }

     if (LOG.isInfoEnabled()) {

       LOG.info("aggregateWorkerStats: Aggregation found " + globalStats +

           " on superstep = " + getSuperstep());

     }

     return globalStats;

   }

   /**

    * Write superstep metrics to own file in HDFS

    * @param superstep the current superstep

    * @param aggregatedMetrics the aggregated metrics to write

    */

   private void printAggregatedMetricsToHDFS(

       long superstep, AggregatedMetrics aggregatedMetrics) {

     ImmutableClassesGiraphConfiguration conf = getConfiguration();

     PrintStream out = null;

     Path dir = new Path(GiraphConstants.METRICS_DIRECTORY.get(conf));

     Path outFile = new Path(GiraphConstants.METRICS_DIRECTORY.get(conf) +

         Path.SEPARATOR_CHAR + "superstep_" + superstep + ".metrics");

     try {

       FileSystem fs;

       fs = FileSystem.get(conf);

       if (!fs.exists(dir)) {

         fs.mkdirs(dir);

       }

       if (fs.exists(outFile)) {

         throw new RuntimeException(

             "printAggregatedMetricsToHDFS: metrics file exists");

       }

       out = new PrintStream(fs.create(outFile), false,

           Charset.defaultCharset().name());

       aggregatedMetrics.print(superstep, out);

     } catch (IOException e) {

       throw new RuntimeException(

           "printAggregatedMetricsToHDFS: error creating metrics file", e);

     } finally {

       if (out != null) {

         out.close();

       }

     }

   }

   /**

    * Finalize the checkpoint file prefixes by taking the chosen workers and

    * writing them to a finalized file.  Also write out the master

    * aggregated aggregator array from the previous superstep.

    *

    * @param superstep superstep to finalize

    * @param chosenWorkerInfoList list of chosen workers that will be finalized

    * @throws IOException

    * @throws InterruptedException

    * @throws KeeperException

    */

   private void finalizeCheckpoint(long superstep,

     List<WorkerInfo> chosenWorkerInfoList)

     throws IOException, KeeperException, InterruptedException {

     Path finalizedCheckpointPath =

         new Path(getCheckpointBasePath(superstep) +

             CheckpointingUtils.CHECKPOINT_FINALIZED_POSTFIX);

     try {

       getFs().delete(finalizedCheckpointPath, false);

     } catch (IOException e) {

       LOG.warn("finalizedValidCheckpointPrefixes: Removed old file " +

           finalizedCheckpointPath);

     }

     // Format:

     // <global statistics>

     // <superstep classes>

     // <number of files>

     // <used file prefix 0><used file prefix 1>...

     // <aggregator data>

     // <masterCompute data>

     FSDataOutputStream finalizedOutputStream =

         getFs().create(finalizedCheckpointPath);

     String superstepFinishedNode =

         getSuperstepFinishedPath(getApplicationAttempt(), superstep - 1);

     finalizedOutputStream.write(

         getZkExt().getData(superstepFinishedNode, false, null));

     finalizedOutputStream.writeInt(chosenWorkerInfoList.size());

     finalizedOutputStream.writeUTF(getCheckpointBasePath(superstep));

     for (WorkerInfo chosenWorkerInfo : chosenWorkerInfoList) {

       finalizedOutputStream.writeInt(getWorkerId(chosenWorkerInfo));

     }

     globalCommHandler.getAggregatorHandler().write(finalizedOutputStream);

     aggregatorTranslation.write(finalizedOutputStream);

     masterCompute.write(finalizedOutputStream);

     finalizedOutputStream.close();

     lastCheckpointedSuperstep = superstep;

     GiraphStats.getInstance().

         getLastCheckpointedSuperstep().setValue(superstep);

   }

   /**

    * Assign the partitions for this superstep.  If there are changes,

    * the workers will know how to do the exchange.  If this was a restarted

    * superstep, then make sure to provide information on where to find the

    * checkpoint file.

    */

   private void assignPartitionOwners() {

     Collection<PartitionOwner> partitionOwners;

     if (getSuperstep() == INPUT_SUPERSTEP) {

       partitionOwners =

           masterGraphPartitioner.createInitialPartitionOwners(

               chosenWorkerInfoList, maxWorkers);

       if (partitionOwners.isEmpty()) {

         throw new IllegalStateException(

             "assignAndExchangePartitions: No partition owners set");

       }

     } else if (getRestartedSuperstep() == getSuperstep()) {

       // If restarted, prepare the checkpoint restart

       try {

         partitionOwners = prepareCheckpointRestart(getSuperstep());

       } catch (IOException e) {

         throw new IllegalStateException(

             "assignPartitionOwners: IOException on preparing", e);

       } catch (KeeperException e) {

         throw new IllegalStateException(

             "assignPartitionOwners: KeeperException on preparing", e);

       } catch (InterruptedException e) {

         throw new IllegalStateException(

             "assignPartitionOwners: InteruptedException on preparing",

             e);

       }

       masterGraphPartitioner.setPartitionOwners(partitionOwners);

     } else {

       partitionOwners =

           masterGraphPartitioner.generateChangedPartitionOwners(

               allPartitionStatsList,

               chosenWorkerInfoList,

               maxWorkers,

               getSuperstep());

       PartitionUtils.analyzePartitionStats(partitionOwners,

           allPartitionStatsList);

     }

     checkPartitions(masterGraphPartitioner.getCurrentPartitionOwners());

     // There will be some exchange of partitions

     if (!partitionOwners.isEmpty()) {

       String vertexExchangePath =

           getPartitionExchangePath(getApplicationAttempt(),

               getSuperstep());

       try {

         getZkExt().createOnceExt(vertexExchangePath,

             null,

             Ids.OPEN_ACL_UNSAFE,

             CreateMode.PERSISTENT,

             true);

       } catch (KeeperException e) {

         throw new IllegalStateException(

             "assignPartitionOwners: KeeperException creating " +

                 vertexExchangePath);

       } catch (InterruptedException e) {

         throw new IllegalStateException(

             "assignPartitionOwners: InterruptedException creating " +

                 vertexExchangePath);

       }

     }

     AddressesAndPartitionsWritable addressesAndPartitions =

         new AddressesAndPartitionsWritable(masterInfo, chosenWorkerInfoList,

             partitionOwners);

     // Send assignments to every worker

     // TODO for very large number of partitions we might want to split this

     // across multiple requests

     for (WorkerInfo workerInfo : chosenWorkerInfoList) {

       masterClient.sendWritableRequest(workerInfo.getTaskId(),

           new AddressesAndPartitionsRequest(addressesAndPartitions));

     }

   }

   /**

    * Check if partition ids are valid

    *

    * @param partitionOwners List of partition ids for current superstep

    */

   private void checkPartitions(Collection<PartitionOwner> partitionOwners) {

     for (PartitionOwner partitionOwner : partitionOwners) {

       int partitionId = partitionOwner.getPartitionId();

       if (partitionId < 0 || partitionId >= partitionOwners.size()) {

         throw new IllegalStateException("checkPartitions: " +

             "Invalid partition id " + partitionId +

             " - partition ids must be values from 0 to (numPartitions - 1)");

       }

     }

   }

   /**

    * Check whether the workers chosen for this superstep are still alive

    *

    * @param chosenWorkerInfoHealthPath Path to the healthy workers in ZooKeeper

    * @param chosenWorkerInfoList List of the healthy workers

    * @return a list of dead workers. Empty list if all workers are alive.

    * @throws InterruptedException

    * @throws KeeperException

    */

   private Collection<WorkerInfo> superstepChosenWorkerAlive(

     String chosenWorkerInfoHealthPath,

     List<WorkerInfo> chosenWorkerInfoList)

     throws KeeperException, InterruptedException {

     List<WorkerInfo> chosenWorkerInfoHealthyList =

         getWorkerInfosFromPath(chosenWorkerInfoHealthPath, false);

     Set<WorkerInfo> chosenWorkerInfoHealthySet =

         new HashSet<WorkerInfo>(chosenWorkerInfoHealthyList);

     List<WorkerInfo> deadWorkers = new ArrayList<>();

     for (WorkerInfo chosenWorkerInfo : chosenWorkerInfoList) {

       if (!chosenWorkerInfoHealthySet.contains(chosenWorkerInfo)) {

         deadWorkers.add(chosenWorkerInfo);

       }

     }

     return deadWorkers;

   }

   @Override

   public void restartFromCheckpoint(long checkpoint) {

     // Process:

     // 1. Increase the application attempt and set to the correct checkpoint

     // 2. Send command to all workers to restart their tasks

     setApplicationAttempt(getApplicationAttempt() + 1);

     setCachedSuperstep(checkpoint);

     setRestartedSuperstep(checkpoint);

     checkpointStatus = CheckpointStatus.NONE;

     setJobState(ApplicationState.START_SUPERSTEP,

         getApplicationAttempt(),

         checkpoint);

   }

   /**

    * Safely removes node from zookeeper.

    * Ignores if node is already removed. Can only throw runtime exception if

    * anything wrong.

    * @param path path to the node to be removed.

    */

   private void zkDeleteNode(String path) {

     try {

       getZkExt().deleteExt(path, -1, true);

     } catch (KeeperException.NoNodeException e) {

       LOG.info("zkDeleteNode: node has already been removed " + path);

     } catch (InterruptedException e) {

       throw new RuntimeException(

           "zkDeleteNode: InterruptedException", e);

     } catch (KeeperException e) {

       throw new RuntimeException(

           "zkDeleteNode: KeeperException", e);

     }

   }

   @Override

   public long getLastGoodCheckpoint() throws IOException {

     // Find the last good checkpoint if none have been written to the

     // knowledge of this master