/*
    * 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.conf;
  
  import io.netty.buffer.ByteBufAllocator;
  import io.netty.buffer.PooledByteBufAllocator;
  import io.netty.buffer.UnpooledByteBufAllocator;
  
  import java.net.InetAddress;
  import java.net.UnknownHostException;
  
  import org.apache.giraph.aggregators.AggregatorWriter;
  import org.apache.giraph.bsp.checkpoints.CheckpointSupportedChecker;
  import org.apache.giraph.combiner.MessageCombiner;
  import org.apache.giraph.edge.OutEdges;
  import org.apache.giraph.edge.ReuseObjectsOutEdges;
  import org.apache.giraph.factories.ComputationFactory;
  import org.apache.giraph.factories.VertexValueFactory;
  import org.apache.giraph.graph.Computation;
  import org.apache.giraph.graph.MapperObserver;
  import org.apache.giraph.graph.Vertex;
  import org.apache.giraph.graph.VertexResolver;
  import org.apache.giraph.graph.VertexValueCombiner;
  import org.apache.giraph.io.EdgeInputFormat;
  import org.apache.giraph.io.EdgeOutputFormat;
  import org.apache.giraph.io.MappingInputFormat;
  import org.apache.giraph.io.VertexInputFormat;
  import org.apache.giraph.io.VertexOutputFormat;
  import org.apache.giraph.io.filters.EdgeInputFilter;
  import org.apache.giraph.io.filters.VertexInputFilter;
  import org.apache.giraph.job.GiraphJobObserver;
  import org.apache.giraph.job.GiraphJobRetryChecker;
  import org.apache.giraph.master.MasterCompute;
  import org.apache.giraph.master.MasterObserver;
  import org.apache.giraph.partition.GraphPartitionerFactory;
  import org.apache.giraph.partition.Partition;
  import org.apache.giraph.partition.ReusesObjectsPartition;
  import org.apache.giraph.utils.GcObserver;
  import org.apache.giraph.utils.ReflectionUtils;
  import org.apache.giraph.worker.WorkerContext;
  import org.apache.giraph.worker.WorkerObserver;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.mapreduce.Mapper;
  import org.apache.hadoop.net.DNS;
  
  /**
   * Adds user methods specific to Giraph.  This will be put into an
   * ImmutableClassesGiraphConfiguration that provides the configuration plus
   * the immutable classes.
   *
   * Keeps track of parameters which were set so it easily set them in another
   * copy of configuration.
   */
  public class GiraphConfiguration extends Configuration
      implements GiraphConstants {
    /** ByteBufAllocator to be used by netty */
    private ByteBufAllocator nettyBufferAllocator = null;
  
    /**
     * Constructor that creates the configuration
     */
    public GiraphConfiguration() {
      configureHadoopSecurity();
    }
  
    /**
     * Constructor.
     *
     * @param conf Configuration
     */
    public GiraphConfiguration(Configuration conf) {
      super(conf);
      configureHadoopSecurity();
    }
  
    /**
     * Get name of computation being run. We leave this up to the
     * {@link ComputationFactory} to decide what to return.
     *
     * @return Name of computation being run
     */
    public String getComputationName() {
      ComputationFactory compFactory = ReflectionUtils.newInstance(
         getComputationFactoryClass());
     return compFactory.computationName(this);
   }
 
   /**
    * Get the user's subclassed {@link ComputationFactory}
    *
    * @return User's computation factory class
    */
   public Class<? extends ComputationFactory> getComputationFactoryClass() {
     return COMPUTATION_FACTORY_CLASS.get(this);
   }
 
   /**
    * Get the user's subclassed {@link Computation}
    *
    * @return User's computation class
    */
   public Class<? extends Computation> getComputationClass() {
     return COMPUTATION_CLASS.get(this);
   }
 
   /**
    * Set the computation class (required)
    *
    * @param computationClass Runs vertex computation
    */
   public void setComputationClass(
       Class<? extends Computation> computationClass) {
     COMPUTATION_CLASS.set(this, computationClass);
   }
 
   /**
    * Set the vertex value factory class
    *
    * @param vertexValueFactoryClass Creates default vertex values
    */
   public final void setVertexValueFactoryClass(
       Class<? extends VertexValueFactory> vertexValueFactoryClass) {
     VERTEX_VALUE_FACTORY_CLASS.set(this, vertexValueFactoryClass);
   }
 
   /**
    * Set the edge input filter class
    *
    * @param edgeFilterClass class to use
    */
   public void setEdgeInputFilterClass(
       Class<? extends EdgeInputFilter> edgeFilterClass) {
     EDGE_INPUT_FILTER_CLASS.set(this, edgeFilterClass);
   }
 
   /**
    * Set the vertex input filter class
    *
    * @param vertexFilterClass class to use
    */
   public void setVertexInputFilterClass(
       Class<? extends VertexInputFilter> vertexFilterClass) {
     VERTEX_INPUT_FILTER_CLASS.set(this, vertexFilterClass);
   }
 
   /**
    * Get the vertex edges class
    *
    * @return vertex edges class
    */
   public Class<? extends OutEdges> getOutEdgesClass() {
     return VERTEX_EDGES_CLASS.get(this);
   }
 
   /**
    * Set the vertex edges class
    *
    * @param outEdgesClass Determines the way edges are stored
    */
   public final void setOutEdgesClass(
       Class<? extends OutEdges> outEdgesClass) {
     VERTEX_EDGES_CLASS.set(this, outEdgesClass);
   }
 
   /**
    * Set the vertex implementation class
    *
    * @param vertexClass class of the vertex implementation
    */
   public final void setVertexClass(Class<? extends Vertex> vertexClass) {
     VERTEX_CLASS.set(this, vertexClass);
   }
 
 
   /**
    * Set the vertex edges class used during edge-based input (if different
    * from the one used during computation)
    *
    * @param inputOutEdgesClass Determines the way edges are stored
    */
   public final void setInputOutEdgesClass(
       Class<? extends OutEdges> inputOutEdgesClass) {
     INPUT_VERTEX_EDGES_CLASS.set(this, inputOutEdgesClass);
   }
 
   /**
    * True if the {@link org.apache.giraph.edge.OutEdges} implementation
    * copies the passed edges to its own data structure,
    * i.e. it doesn't keep references to Edge objects, target vertex ids or edge
    * values passed to add() or initialize().
    * This makes it possible to reuse edge objects passed to the data
    * structure, to minimize object instantiation (see for example
    * EdgeStore#addPartitionEdges()).
    *
    * @return True iff we can reuse the edge objects
    */
   public boolean reuseEdgeObjects() {
     return ReuseObjectsOutEdges.class.isAssignableFrom(
         getOutEdgesClass());
   }
 
   /**
    * True if the {@link Partition} implementation copies the passed vertices
    * to its own data structure, i.e. it doesn't keep references to Vertex
    * objects passed to it.
    * This makes it possible to reuse vertex objects passed to the data
    * structure, to minimize object instantiation.
    *
    * @return True iff we can reuse the vertex objects
    */
   public boolean reuseVertexObjects() {
     return ReusesObjectsPartition.class.isAssignableFrom(getPartitionClass());
   }
 
   /**
    * Get Partition class used
    * @return Partition class
    */
   public Class<? extends Partition> getPartitionClass() {
     return PARTITION_CLASS.get(this);
   }
 
   /**
    * Does the job have a {@link VertexInputFormat}?
    *
    * @return True iff a {@link VertexInputFormat} has been specified.
    */
   public boolean hasVertexInputFormat() {
     return VERTEX_INPUT_FORMAT_CLASS.get(this) != null;
   }
 
   /**
    * Set the vertex input format class (required)
    *
    * @param vertexInputFormatClass Determines how graph is input
    */
   public void setVertexInputFormatClass(
       Class<? extends VertexInputFormat> vertexInputFormatClass) {
     VERTEX_INPUT_FORMAT_CLASS.set(this, vertexInputFormatClass);
   }
 
   /**
    * Does the job have a {@link EdgeInputFormat}?
    *
    * @return True iff a {@link EdgeInputFormat} has been specified.
    */
   public boolean hasEdgeInputFormat() {
     return EDGE_INPUT_FORMAT_CLASS.get(this) != null;
   }
 
   /**
    * Set the edge input format class (required)
    *
    * @param edgeInputFormatClass Determines how graph is input
    */
   public void setEdgeInputFormatClass(
       Class<? extends EdgeInputFormat> edgeInputFormatClass) {
     EDGE_INPUT_FORMAT_CLASS.set(this, edgeInputFormatClass);
   }
 
   /**
    * Set the mapping input format class (optional)
    *
    * @param mappingInputFormatClass Determines how mappings are input
    */
   public void setMappingInputFormatClass(
     Class<? extends MappingInputFormat> mappingInputFormatClass) {
     MAPPING_INPUT_FORMAT_CLASS.set(this, mappingInputFormatClass);
   }
 
   /**
    * Set the master class (optional)
    *
    * @param masterComputeClass Runs master computation
    */
   public final void setMasterComputeClass(
       Class<? extends MasterCompute> masterComputeClass) {
     MASTER_COMPUTE_CLASS.set(this, masterComputeClass);
   }
 
   /**
    * Add a MasterObserver class (optional)
    *
    * @param masterObserverClass MasterObserver class to add.
    */
   public final void addMasterObserverClass(
       Class<? extends MasterObserver> masterObserverClass) {
     MASTER_OBSERVER_CLASSES.add(this, masterObserverClass);
   }
 
   /**
    * Add a WorkerObserver class (optional)
    *
    * @param workerObserverClass WorkerObserver class to add.
    */
   public final void addWorkerObserverClass(
       Class<? extends WorkerObserver> workerObserverClass) {
     WORKER_OBSERVER_CLASSES.add(this, workerObserverClass);
   }
 
   /**
    * Add a MapperObserver class (optional)
    *
    * @param mapperObserverClass MapperObserver class to add.
    */
   public final void addMapperObserverClass(
       Class<? extends MapperObserver> mapperObserverClass) {
     MAPPER_OBSERVER_CLASSES.add(this, mapperObserverClass);
   }
 
   /**
    * Add a GcObserver class (optional)
    *
    * @param gcObserverClass GcObserver class to add.
    */
   public final void addGcObserverClass(
       Class<? extends GcObserver> gcObserverClass) {
     GC_OBSERVER_CLASSES.add(this, gcObserverClass);
   }
 
   /**
    * Get job observer class
    *
    * @return GiraphJobObserver class set.
    */
   public Class<? extends GiraphJobObserver> getJobObserverClass() {
     return JOB_OBSERVER_CLASS.get(this);
   }
 
   /**
    * Set job observer class
    *
    * @param klass GiraphJobObserver class to set.
    */
   public void setJobObserverClass(Class<? extends GiraphJobObserver> klass) {
     JOB_OBSERVER_CLASS.set(this, klass);
   }
 
   /**
    * Get job retry checker class
    *
    * @return GiraphJobRetryChecker class set.
    */
   public Class<? extends GiraphJobRetryChecker> getJobRetryCheckerClass() {
     return JOB_RETRY_CHECKER_CLASS.get(this);
   }
 
   /**
    * Set job retry checker class
    *
    * @param klass GiraphJobRetryChecker class to set.
    */
   public void setJobRetryCheckerClass(
       Class<? extends GiraphJobRetryChecker> klass) {
     JOB_RETRY_CHECKER_CLASS.set(this, klass);
   }
 
   /**
    * Check whether to enable jmap dumping thread.
    *
    * @return true if jmap dumper is enabled.
    */
   public boolean isJMapHistogramDumpEnabled() {
     return JMAP_ENABLE.get(this);
   }
 
   /**
    * Check whether to enable heap memory supervisor thread
    *
    * @return true if jmap dumper is reactively enabled
    */
   public boolean isReactiveJmapHistogramDumpEnabled() {
     return REACTIVE_JMAP_ENABLE.get(this);
   }
 
   /**
    * Set mapping from a key name to a list of classes.
    *
    * @param name String key name to use.
    * @param xface interface of the classes being set.
    * @param klasses Classes to set.
    */
   public final void setClasses(String name, Class<?> xface,
                                Class<?> ... klasses) {
     String[] klassNames = new String[klasses.length];
     for (int i = 0; i < klasses.length; ++i) {
       Class<?> klass = klasses[i];
       if (!xface.isAssignableFrom(klass)) {
         throw new RuntimeException(klass + " does not implement " +
             xface.getName());
       }
       klassNames[i] = klasses[i].getName();
     }
     setStrings(name, klassNames);
   }
 
   /**
    * Does the job have a {@link VertexOutputFormat}?
    *
    * @return True iff a {@link VertexOutputFormat} has been specified.
    */
   public boolean hasVertexOutputFormat() {
     return VERTEX_OUTPUT_FORMAT_CLASS.get(this) != null;
   }
 
   /**
    * Set the vertex output format class (optional)
    *
    * @param vertexOutputFormatClass Determines how graph is output
    */
   public final void setVertexOutputFormatClass(
       Class<? extends VertexOutputFormat> vertexOutputFormatClass) {
     VERTEX_OUTPUT_FORMAT_CLASS.set(this, vertexOutputFormatClass);
   }
 
 
   /**
    * Does the job have a {@link EdgeOutputFormat} subdir?
    *
    * @return True iff a {@link EdgeOutputFormat} subdir has been specified.
    */
   public boolean hasVertexOutputFormatSubdir() {
     return !VERTEX_OUTPUT_FORMAT_SUBDIR.get(this).isEmpty();
   }
 
   /**
    * Set the vertex output format path
    *
    * @param path path where the verteces will be written
    */
   public final void setVertexOutputFormatSubdir(String path) {
     VERTEX_OUTPUT_FORMAT_SUBDIR.set(this, path);
   }
 
   /**
    * Check if output should be done during computation
    *
    * @return True iff output should be done during computation
    */
   public final boolean doOutputDuringComputation() {
     return DO_OUTPUT_DURING_COMPUTATION.get(this);
   }
 
   /**
    * Set whether or not we should do output during computation
    *
    * @param doOutputDuringComputation True iff we want output to happen
    *                                  during computation
    */
   public final void setDoOutputDuringComputation(
       boolean doOutputDuringComputation) {
     DO_OUTPUT_DURING_COMPUTATION.set(this, doOutputDuringComputation);
   }
 
   /**
    * Check if VertexOutputFormat is thread-safe
    *
    * @return True iff VertexOutputFormat is thread-safe
    */
   public final boolean vertexOutputFormatThreadSafe() {
     return VERTEX_OUTPUT_FORMAT_THREAD_SAFE.get(this);
   }
 
   /**
    * Set whether or not selected VertexOutputFormat is thread-safe
    *
    * @param vertexOutputFormatThreadSafe True iff selected VertexOutputFormat
    *                                     is thread-safe
    */
   public final void setVertexOutputFormatThreadSafe(
       boolean vertexOutputFormatThreadSafe) {
     VERTEX_OUTPUT_FORMAT_THREAD_SAFE.set(this, vertexOutputFormatThreadSafe);
   }
 
   /**
    * Does the job have a {@link EdgeOutputFormat}?
    *
    * @return True iff a {@link EdgeOutputFormat} has been specified.
    */
   public boolean hasEdgeOutputFormat() {
     return EDGE_OUTPUT_FORMAT_CLASS.get(this) != null;
   }
 
   /**
    * Set the edge output format class (optional)
    *
    * @param edgeOutputFormatClass Determines how graph is output
    */
   public final void setEdgeOutputFormatClass(
       Class<? extends EdgeOutputFormat> edgeOutputFormatClass) {
     EDGE_OUTPUT_FORMAT_CLASS.set(this, edgeOutputFormatClass);
   }
 
   /**
    * Does the job have a {@link EdgeOutputFormat} subdir?
    *
    * @return True iff a {@link EdgeOutputFormat} subdir has been specified.
    */
   public boolean hasEdgeOutputFormatSubdir() {
     return !EDGE_OUTPUT_FORMAT_SUBDIR.get(this).isEmpty();
   }
 
   /**
    * Set the edge output format path
    *
    * @param path path where the edges will be written
    */
   public final void setEdgeOutputFormatSubdir(String path) {
     EDGE_OUTPUT_FORMAT_SUBDIR.set(this, path);
   }
 
   /**
    * Get the number of threads to use for writing output in the end of the
    * application. If output format is not thread safe, returns 1.
    *
    * @return Number of output threads
    */
   public final int getNumOutputThreads() {
     if (!vertexOutputFormatThreadSafe()) {
       return 1;
     } else {
       return NUM_OUTPUT_THREADS.get(this);
     }
   }
 
   /**
    * Set the number of threads to use for writing output in the end of the
    * application. Will be used only if {#vertexOutputFormatThreadSafe} is true.
    *
    * @param numOutputThreads Number of output threads
    */
   public void setNumOutputThreads(int numOutputThreads) {
     NUM_OUTPUT_THREADS.set(this, numOutputThreads);
   }
 
   /**
    * Set the message combiner class (optional)
    *
    * @param messageCombinerClass Determines how vertex messages are combined
    */
   public void setMessageCombinerClass(
       Class<? extends MessageCombiner> messageCombinerClass) {
     MESSAGE_COMBINER_CLASS.set(this, messageCombinerClass);
   }
 
   /**
    * Set the graph partitioner class (optional)
    *
    * @param graphPartitionerFactoryClass Determines how the graph is partitioned
    */
   public final void setGraphPartitionerFactoryClass(
       Class<? extends GraphPartitionerFactory> graphPartitionerFactoryClass) {
     GRAPH_PARTITIONER_FACTORY_CLASS.set(this, graphPartitionerFactoryClass);
   }
 
   /**
    * Set the vertex resolver class (optional)
    *
    * @param vertexResolverClass Determines how vertex mutations are resolved
    */
   public final void setVertexResolverClass(
       Class<? extends VertexResolver> vertexResolverClass) {
     VERTEX_RESOLVER_CLASS.set(this, vertexResolverClass);
   }
 
   /**
    * Whether to create a vertex that doesn't exist when it receives messages.
    * This only affects DefaultVertexResolver.
    *
    * @return true if we should create non existent vertices that get messages.
    */
   public final boolean getResolverCreateVertexOnMessages() {
     return RESOLVER_CREATE_VERTEX_ON_MSGS.get(this);
   }
 
   /**
    * Set whether to create non existent vertices when they receive messages.
    *
    * @param v true if we should create vertices when they get messages.
    */
   public final void setResolverCreateVertexOnMessages(boolean v) {
     RESOLVER_CREATE_VERTEX_ON_MSGS.set(this, v);
   }
 
   /**
    * Set the vertex value combiner class (optional)
    *
    * @param vertexValueCombinerClass Determines how vertices are combined
    */
   public final void setVertexValueCombinerClass(
       Class<? extends VertexValueCombiner> vertexValueCombinerClass) {
     VERTEX_VALUE_COMBINER_CLASS.set(this, vertexValueCombinerClass);
   }
 
   /**
    * Set the worker context class (optional)
    *
    * @param workerContextClass Determines what code is executed on a each
    *        worker before and after each superstep and computation
    */
   public final void setWorkerContextClass(
       Class<? extends WorkerContext> workerContextClass) {
     WORKER_CONTEXT_CLASS.set(this, workerContextClass);
   }
 
   /**
    * Set the aggregator writer class (optional)
    *
    * @param aggregatorWriterClass Determines how the aggregators are
    *        written to file at the end of the job
    */
   public final void setAggregatorWriterClass(
       Class<? extends AggregatorWriter> aggregatorWriterClass) {
     AGGREGATOR_WRITER_CLASS.set(this, aggregatorWriterClass);
   }
 
   /**
    * Set the partition class (optional)
    *
    * @param partitionClass Determines the why partitions are stored
    */
   public final void setPartitionClass(
       Class<? extends Partition> partitionClass) {
     PARTITION_CLASS.set(this, partitionClass);
   }
 
   /**
    * Set worker configuration for determining what is required for
    * a superstep.
    *
    * @param minWorkers Minimum workers to do a superstep
    * @param maxWorkers Maximum workers to do a superstep
    *        (max map tasks in job)
    * @param minPercentResponded 0 - 100 % of the workers required to
    *        have responded before continuing the superstep
    */
   public final void setWorkerConfiguration(int minWorkers,
                                            int maxWorkers,
                                            float minPercentResponded) {
     setInt(MIN_WORKERS, minWorkers);
     setInt(MAX_WORKERS, maxWorkers);
     MIN_PERCENT_RESPONDED.set(this, minPercentResponded);
   }
 
   public final int getMinWorkers() {
     return getInt(MIN_WORKERS, -1);
   }
 
   public final int getMaxWorkers() {
     return getInt(MAX_WORKERS, -1);
   }
 
   public final float getMinPercentResponded() {
     return MIN_PERCENT_RESPONDED.get(this);
   }
 
   /**
    * How many mappers is job asking for, taking into account whether master
    * is running on the same mapper as worker or not
    *
    * @return How many mappers is job asking for
    */
   public final int getMaxMappers() {
     return getMaxWorkers() + (SPLIT_MASTER_WORKER.get(this) ? 1 : 0);
   }
 
   /**
    * Utilize an existing ZooKeeper service.  If this is not set, ZooKeeper
    * will be dynamically started by Giraph for this job.
    *
    * @param serverList Comma separated list of servers and ports
    *        (i.e. zk1:2221,zk2:2221)
    */
   public final void setZooKeeperConfiguration(String serverList) {
     ZOOKEEPER_LIST.set(this, serverList);
   }
 
   /**
    * Getter for SPLIT_MASTER_WORKER flag.
    *
    * @return boolean flag value.
    */
   public final boolean getSplitMasterWorker() {
     return SPLIT_MASTER_WORKER.get(this);
   }
 
   /**
    * Get array of MasterObserver classes set in the configuration.
    *
    * @return array of MasterObserver classes.
    */
   public Class<? extends MasterObserver>[] getMasterObserverClasses() {
     return MASTER_OBSERVER_CLASSES.getArray(this);
   }
 
   /**
    * Get array of WorkerObserver classes set in configuration.
    *
    * @return array of WorkerObserver classes.
    */
   public Class<? extends WorkerObserver>[] getWorkerObserverClasses() {
     return WORKER_OBSERVER_CLASSES.getArray(this);
   }
 
   /**
    * Get array of MapperObserver classes set in configuration.
    *
    * @return array of MapperObserver classes.
    */
   public Class<? extends MapperObserver>[] getMapperObserverClasses() {
     return MAPPER_OBSERVER_CLASSES.getArray(this);
   }
 
   /**
    * Get array of GcObserver classes set in configuration.
    *
    * @return array of GcObserver classes.
    */
   public Class<? extends GcObserver>[] getGcObserverClasses() {
     return GC_OBSERVER_CLASSES.getArray(this);
   }
 
   /**
    * Whether to track, print, and aggregate metrics.
    *
    * @return true if metrics are enabled, false otherwise (default)
    */
   public boolean metricsEnabled() {
     return METRICS_ENABLE.isTrue(this);
   }
 
   /**
    * Get the task partition
    *
    * @return The task partition or -1 if not set
    */
   public int getTaskPartition() {
     return getInt("mapred.task.partition", -1);
   }
 
   /**
    * Is this a "pure YARN" Giraph job, or is a MapReduce layer (v1 or v2)
    * actually managing our cluster nodes, i.e. each task is a Mapper.
    *
    * @return TRUE if this is a pure YARN job.
    */
   public boolean isPureYarnJob() {
     return IS_PURE_YARN_JOB.get(this);
   }
 
   /**
    * Jars required in "Pure YARN" jobs (names only, no paths) should
    * be listed here in full, including Giraph framework jar(s).
    *
    * @return the comma-separated list of jar names for export to cluster.
    */
   public String getYarnLibJars() {
     return GIRAPH_YARN_LIBJARS.get(this);
   }
 
   /**
    * Populate jar list for Pure YARN jobs.
    *
    * @param jarList a comma-separated list of jar names
    */
   public void setYarnLibJars(String jarList) {
     GIRAPH_YARN_LIBJARS.set(this, jarList);
   }
 
   /**
    * Get heap size (in MB) for each task in our Giraph job run,
    * assuming this job will run on the "pure YARN" profile.
    *
    * @return the heap size for all tasks, in MB
    */
   public int getYarnTaskHeapMb() {
     return GIRAPH_YARN_TASK_HEAP_MB.get(this);
   }
 
   /**
    * Set heap size for Giraph tasks in our job run, assuming
    * the job will run on the "pure YARN" profile.
    *
    * @param heapMb the heap size for all tasks
    */
   public void setYarnTaskHeapMb(int heapMb) {
     GIRAPH_YARN_TASK_HEAP_MB.set(this, heapMb);
   }
 
   /**
    * Get the ZooKeeper list.
    *
    * @return ZooKeeper list of strings, comma separated or null if none set.
    */
   public String getZookeeperList() {
     return ZOOKEEPER_LIST.get(this);
   }
 
   /**
    * Set the ZooKeeper list to the provided list. This method is used when the
    * ZooKeeper is started internally and will set the zkIsExternal option to
    * false as well.
    *
    * @param zkList list of strings, comma separated of zookeeper servers
    */
   public void setZookeeperList(String zkList) {
     ZOOKEEPER_LIST.set(this, zkList);
     ZOOKEEPER_IS_EXTERNAL.set(this, false);
   }
 
   /**
    * Was ZooKeeper provided externally?
    *
    * @return true iff was zookeeper is external
    */
   public boolean isZookeeperExternal() {
     return ZOOKEEPER_IS_EXTERNAL.get(this);
   }
 
   public String getLocalLevel() {
     return LOG_LEVEL.get(this);
   }
 
   /**
    * Use the log thread layout option?
    *
    * @return True if use the log thread layout option, false otherwise
    */
   public boolean useLogThreadLayout() {
     return LOG_THREAD_LAYOUT.get(this);
   }
 
   /**
    * is this job run a local test?
    *
    * @return the test status as recorded in the Configuration
    */
   public boolean getLocalTestMode() {
     return LOCAL_TEST_MODE.get(this);
   }
 
   /**
    * Flag this job as a local test run.
    *
    * @param flag the test status for this job
    */
   public void setLocalTestMode(boolean flag) {
     LOCAL_TEST_MODE.set(this, flag);
   }
 
   public int getZooKeeperSessionTimeout() {
     return ZOOKEEPER_SESSION_TIMEOUT.get(this);
   }
 
   public int getZookeeperOpsMaxAttempts() {
     return ZOOKEEPER_OPS_MAX_ATTEMPTS.get(this);
   }
 
   public int getZookeeperOpsRetryWaitMsecs() {
     return ZOOKEEPER_OPS_RETRY_WAIT_MSECS.get(this);
   }
 
   public boolean getNettyServerUseExecutionHandler() {
     return NETTY_SERVER_USE_EXECUTION_HANDLER.get(this);
   }
 
   public int getNettyServerThreads() {
     return NETTY_SERVER_THREADS.get(this);
   }
 
   public int getNettyServerExecutionThreads() {
     return NETTY_SERVER_EXECUTION_THREADS.get(this);
   }
 
   /**
    * Get the netty server execution concurrency.  This depends on whether the
    * netty server execution handler exists.
    *
    * @return Server concurrency
    */
   public int getNettyServerExecutionConcurrency() {
     if (getNettyServerUseExecutionHandler()) {
       return getNettyServerExecutionThreads();
     } else {
       return getNettyServerThreads();
     }
   }
 
   /**
    * Used by netty client and server to create ByteBufAllocator
    *
    * @return ByteBufAllocator
    */
   public ByteBufAllocator getNettyAllocator() {
     if (nettyBufferAllocator == null) {
       if (NETTY_USE_POOLED_ALLOCATOR.get(this)) { // Use pooled allocator
         nettyBufferAllocator = new PooledByteBufAllocator(
           NETTY_USE_DIRECT_MEMORY.get(this));
       } else { // Use un-pooled allocator
         // Note: Current default settings create un-pooled heap allocator
         nettyBufferAllocator = new UnpooledByteBufAllocator(
             NETTY_USE_DIRECT_MEMORY.get(this));
       }
     }
     return nettyBufferAllocator;
   }
 
   public int getZookeeperConnectionAttempts() {
     return ZOOKEEPER_CONNECTION_ATTEMPTS.get(this);
   }
 
   public int getZooKeeperMinSessionTimeout() {
     return ZOOKEEPER_MIN_SESSION_TIMEOUT.get(this);
   }
 
   public int getZooKeeperMaxSessionTimeout() {
     return ZOOKEEPER_MAX_SESSION_TIMEOUT.get(this);
   }
 
   /**
    * Get the number of map tasks in this job
    *
    * @return Number of map tasks in this job
    */
   public int getMapTasks() {
     int mapTasks = getInt("mapred.map.tasks", -1);
     if (mapTasks == -1) {
       throw new IllegalStateException("getMapTasks: Failed to get the map " +
           "tasks!");
     }
     return mapTasks;
   }
 
   /**
    * Use authentication? (if supported)
    *
    * @return True if should authenticate, false otherwise
    */
   public boolean authenticate() {
     return AUTHENTICATE.get(this);
   }
 
   /**
    * Set the number of compute threads
    *
    * @param numComputeThreads Number of compute threads to use
    */
   public void setNumComputeThreads(int numComputeThreads) {
     NUM_COMPUTE_THREADS.set(this, numComputeThreads);
   }
 
   public int getNumComputeThreads() {
     return NUM_COMPUTE_THREADS.get(this);
   }
 
   /**
    * Set the number of input split threads
    *
    * @param numInputSplitsThreads Number of input split threads to use
    */
   public void setNumInputSplitsThreads(int numInputSplitsThreads) {
     NUM_INPUT_THREADS.set(this, numInputSplitsThreads);
   }
 
   public int getNumInputSplitsThreads() {
     return NUM_INPUT_THREADS.get(this);
   }
 
   public long getInputSplitMaxVertices() {
     return INPUT_SPLIT_MAX_VERTICES.get(this);
   }
 
   public long getInputSplitMaxEdges() {
     return INPUT_SPLIT_MAX_EDGES.get(this);
   }
 
   /**
    * Set whether to use unsafe serialization
    *
    * @param useUnsafeSerialization If true, use unsafe serialization
    */
   public void useUnsafeSerialization(boolean useUnsafeSerialization) {
     USE_UNSAFE_SERIALIZATION.set(this, useUnsafeSerialization);
   }
 
   /**
    * Set the checkpoint frequeuncy of how many supersteps to wait before
    * checkpointing
    *
    * @param checkpointFrequency How often to checkpoint (0 means never)
    */
   public void setCheckpointFrequency(int checkpointFrequency) {
     CHECKPOINT_FREQUENCY.set(this, checkpointFrequency);
   }
 
   /**
    * Get the checkpoint frequeuncy of how many supersteps to wait
    * before checkpointing
    *
    * @return Checkpoint frequency (0 means never)
    */
   public int getCheckpointFrequency() {
     return CHECKPOINT_FREQUENCY.get(this);
   }
 
   /**
    * Check if checkpointing is used
    *
    * @return True iff checkpointing is used
    */
   public boolean useCheckpointing() {
     return getCheckpointFrequency() != 0;
   }
 
   /**
    * Set runtime checkpoint support checker.
    * The instance of this class will have to decide whether
    * checkpointing is allowed on current superstep.
    *
    * @param clazz checkpoint supported checker class
    */
   public void setCheckpointSupportedChecker(
       Class<? extends CheckpointSupportedChecker> clazz) {
     GiraphConstants.CHECKPOINT_SUPPORTED_CHECKER.set(this, clazz);
   }
 
   /**
    * Set the max task attempts
    *
    * @param maxTaskAttempts Max task attempts to use
    */
   public void setMaxTaskAttempts(int maxTaskAttempts) {
     MAX_TASK_ATTEMPTS.set(this, maxTaskAttempts);
   }
 
   /**
    * Get the max task attempts
    *
    * @return Max task attempts or -1, if not set
    */
   public int getMaxTaskAttempts() {
     return MAX_TASK_ATTEMPTS.get(this);
   }
 
   /**
    * Get the number of milliseconds to wait for an event before continuing on
    *
    * @return Number of milliseconds to wait for an event before continuing on
    */
   public int getEventWaitMsecs() {
     return EVENT_WAIT_MSECS.get(this);
   }
 
   /**
    * Set the number of milliseconds to wait for an event before continuing on
    *
    * @param eventWaitMsecs Number of milliseconds to wait for an event before
    *                       continuing on
    */
   public void setEventWaitMsecs(int eventWaitMsecs) {
     EVENT_WAIT_MSECS.set(this, eventWaitMsecs);
   }
 
   /**
    * Get the maximum milliseconds to wait before giving up trying to get the
    * minimum number of workers before a superstep.
    *
    * @return Maximum milliseconds to wait before giving up trying to get the
    *         minimum number of workers before a superstep
    */
   public int getMaxMasterSuperstepWaitMsecs() {
     return MAX_MASTER_SUPERSTEP_WAIT_MSECS.get(this);
   }
 
   public int getMaxCounterWaitMsecs() {
     return MAX_COUNTER_WAIT_MSECS.get(this);
   }
 
   /**
    * Set the maximum milliseconds to wait before giving up trying to get the
    * minimum number of workers before a superstep.
    *
    * @param maxMasterSuperstepWaitMsecs Maximum milliseconds to wait before
    *                                    giving up trying to get the minimum
    *                                    number of workers before a superstep
    */
   public void setMaxMasterSuperstepWaitMsecs(int maxMasterSuperstepWaitMsecs) {
     MAX_MASTER_SUPERSTEP_WAIT_MSECS.set(this, maxMasterSuperstepWaitMsecs);
   }
 
   /**
    * Check environment for Hadoop security token location in case we are
    * executing the Giraph job on a MRv1 Hadoop cluster.
    */
   public void configureHadoopSecurity() {
     String hadoopTokenFilePath = System.getenv("HADOOP_TOKEN_FILE_LOCATION");
     if (hadoopTokenFilePath != null) {
       set("mapreduce.job.credentials.binary", hadoopTokenFilePath);
     }
   }
 
   /**
    * Check if we want to prioritize input splits which reside on the host.
    *
    * @return True iff we want to use input split locality
    */
   public boolean useInputSplitLocality() {
     return USE_INPUT_SPLIT_LOCALITY.get(this);
   }
 
   /**
    * Get the local hostname on the given interface.
    *
    * @return The local hostname
    * @throws UnknownHostException IP address of a host could not be determined
    */
   public String getLocalHostname() throws UnknownHostException {
     return DNS.getDefaultHost(
         GiraphConstants.DNS_INTERFACE.get(this),
         GiraphConstants.DNS_NAMESERVER.get(this)).toLowerCase();
   }
 
   /**
    * Return local host name by default. Or local host IP if preferIP
    * option is set.
    * @return local host name or IP
    * @throws UnknownHostException IP address of a host could not be determined
    */
   public String getLocalHostOrIp() throws UnknownHostException {
     if (GiraphConstants.PREFER_IP_ADDRESSES.get(this)) {
       return InetAddress.getLocalHost().getHostAddress();
     }
     return getLocalHostname();
   }
 
   /**
    * Set the maximum number of supersteps of this application.  After this
    * many supersteps are executed, the application will shutdown.
    *
    * @param maxNumberOfSupersteps Maximum number of supersteps
    */
   public void setMaxNumberOfSupersteps(int maxNumberOfSupersteps) {
     MAX_NUMBER_OF_SUPERSTEPS.set(this, maxNumberOfSupersteps);
   }
 
   /**
    * Get the maximum number of supersteps of this application.  After this
    * many supersteps are executed, the application will shutdown.
    *
    * @return Maximum number of supersteps
    */
   public int getMaxNumberOfSupersteps() {
     return MAX_NUMBER_OF_SUPERSTEPS.get(this);
   }
 
   /**
    * Get the output directory to write YourKit snapshots to
    *
    * @param context Map context
    * @return output directory
    */
   public String getYourKitOutputDir(Mapper.Context context) {
     final String cacheKey = "giraph.yourkit.outputDirCached";
     String outputDir = get(cacheKey);
     if (outputDir == null) {
       outputDir = getStringVars(YOURKIT_OUTPUT_DIR, YOURKIT_OUTPUT_DIR_DEFAULT,
           context);
       set(cacheKey, outputDir);
     }
     return outputDir;
   }
 
   /**
    * Get string, replacing variables in the output.
    *
    * %JOB_ID% =&gt; job id
    * %TASK_ID% =&gt; task id
    * %USER% =&gt; owning user name
    *
    * @param key name of key to lookup
    * @param context mapper context
    * @return value for key, with variables expanded
    */
   public String getStringVars(String key, Mapper.Context context) {
     return getStringVars(key, null, context);
   }
 
   /**
    * Get string, replacing variables in the output.
    *
    * %JOB_ID% =&gt; job id
    * %TASK_ID% =&gt; task id
    * %USER% =&gt; owning user name
    *
    * @param key name of key to lookup
    * @param defaultValue value to return if no mapping exists. This can also
    *                     have variables, which will be substituted.
    * @param context mapper context
    * @return value for key, with variables expanded
    */
   public String getStringVars(String key, String defaultValue,
                               Mapper.Context context) {
     String value = get(key);
     if (value == null) {
       if (defaultValue == null) {
         return null;
       }
       value = defaultValue;
     }
     value = value.replace("%JOB_ID%", context.getJobID().toString());
     value = value.replace("%TASK_ID%", context.getTaskAttemptID().toString());
     value = value.replace("%USER%", get("user.name", "unknown_user"));
     return value;
   }
 
   /**
    * Get option whether to create a source vertex present only in edge input
    *
    * @return CREATE_EDGE_SOURCE_VERTICES option
    */
   public boolean getCreateSourceVertex() {
     return CREATE_EDGE_SOURCE_VERTICES.get(this);
   }
 
   /**
    * set option whether to create a source vertex present only in edge input
    * @param createVertex create source vertex option
    */
   public void setCreateSourceVertex(boolean createVertex) {
     CREATE_EDGE_SOURCE_VERTICES.set(this, createVertex);
   }
 
   /**
    * Get the maximum timeout (in milliseconds) for waiting for all tasks
    * to complete after the job is done.
    *
    * @return Wait task done timeout in milliseconds.
    */
   public int getWaitTaskDoneTimeoutMs() {
     return WAIT_TASK_DONE_TIMEOUT_MS.get(this);
   }
 
   /**
    * Set the maximum timeout (in milliseconds) for waiting for all tasks
    * to complete after the job is done.
    *
    * @param ms Milliseconds to set
    */
   public void setWaitTaskDoneTimeoutMs(int ms) {
     WAIT_TASK_DONE_TIMEOUT_MS.set(this, ms);
   }
 
   /**
    * Check whether to track job progress on client or not
    *
    * @return True if job progress should be tracked on client
    */
   public boolean trackJobProgressOnClient() {
     return TRACK_JOB_PROGRESS_ON_CLIENT.get(this);
   }
 
   /**
    * @return Number of retries when creating an HDFS file before failing.
    */
   public int getHdfsFileCreationRetries() {
     return HDFS_FILE_CREATION_RETRIES.get(this);
   }
 
   /**
    * @return Milliseconds to wait before retrying an HDFS file creation
    *         operation.
    */
   public int getHdfsFileCreationRetryWaitMs() {
     return HDFS_FILE_CREATION_RETRY_WAIT_MS.get(this);
   }
 }