/*

  * 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.handler.codec.ByteToMessageDecoder;

 import io.netty.handler.codec.MessageToByteEncoder;

 import io.netty.handler.codec.compression.JdkZlibDecoder;

 import io.netty.handler.codec.compression.JdkZlibEncoder;

 import io.netty.handler.codec.compression.SnappyFramedDecoder;

 import io.netty.handler.codec.compression.SnappyFramedEncoder;

 import org.apache.giraph.aggregators.AggregatorWriter;

 import org.apache.giraph.combiner.MessageCombiner;

 import org.apache.giraph.comm.messages.MessageEncodeAndStoreType;

 import org.apache.giraph.edge.Edge;

 import org.apache.giraph.edge.EdgeFactory;

 import org.apache.giraph.edge.EdgeStoreFactory;

 import org.apache.giraph.edge.OutEdges;

 import org.apache.giraph.edge.ReusableEdge;

 import org.apache.giraph.factories.ComputationFactory;

 import org.apache.giraph.factories.EdgeValueFactory;

 import org.apache.giraph.factories.MessageValueFactory;

 import org.apache.giraph.factories.OutEdgesFactory;

 import org.apache.giraph.factories.ValueFactories;

 import org.apache.giraph.factories.VertexIdFactory;

 import org.apache.giraph.factories.VertexValueFactory;

 import org.apache.giraph.graph.Computation;

 import org.apache.giraph.graph.Language;

 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.io.internal.WrappedEdgeInputFormat;

 import org.apache.giraph.io.internal.WrappedEdgeOutputFormat;

 import org.apache.giraph.io.internal.WrappedMappingInputFormat;

 import org.apache.giraph.io.internal.WrappedVertexInputFormat;

 import org.apache.giraph.io.internal.WrappedVertexOutputFormat;

 import org.apache.giraph.io.superstep_output.MultiThreadedSuperstepOutput;

 import org.apache.giraph.io.superstep_output.NoOpSuperstepOutput;

 import org.apache.giraph.io.superstep_output.SuperstepOutput;

 import org.apache.giraph.io.superstep_output.SynchronizedSuperstepOutput;

 import org.apache.giraph.job.GiraphJobObserver;

 import org.apache.giraph.job.GiraphJobRetryChecker;

 import org.apache.giraph.mapping.MappingStore;

 import org.apache.giraph.mapping.MappingStoreOps;

 import org.apache.giraph.mapping.translate.TranslateEdge;

 import org.apache.giraph.master.MasterCompute;

 import org.apache.giraph.master.MasterObserver;

 import org.apache.giraph.master.SuperstepClasses;

 import org.apache.giraph.partition.GraphPartitionerFactory;

 import org.apache.giraph.partition.Partition;

 import org.apache.giraph.utils.ExtendedByteArrayDataInput;

 import org.apache.giraph.utils.ExtendedByteArrayDataOutput;

 import org.apache.giraph.utils.ExtendedDataInput;

 import org.apache.giraph.utils.ExtendedDataOutput;

 import org.apache.giraph.utils.GcObserver;

 import org.apache.giraph.utils.ReflectionUtils;

 import org.apache.giraph.utils.UnsafeByteArrayInputStream;

 import org.apache.giraph.utils.UnsafeByteArrayOutputStream;

 import org.apache.giraph.utils.WritableUtils;

 import org.apache.giraph.utils.io.BigDataInputOutput;

 import org.apache.giraph.utils.io.DataInputOutput;

 import org.apache.giraph.utils.io.ExtendedDataInputOutput;

 import org.apache.giraph.worker.WorkerContext;

 import org.apache.giraph.worker.WorkerObserver;

 import org.apache.hadoop.conf.Configuration;

 import org.apache.hadoop.io.NullWritable;

 import org.apache.hadoop.io.Writable;

 import org.apache.hadoop.io.WritableComparable;

 import org.apache.hadoop.mapreduce.Mapper;

 import org.apache.hadoop.util.Progressable;

 import com.google.common.base.Preconditions;

 /**

  * The classes set here are immutable, the remaining configuration is mutable.

  * Classes are immutable and final to provide the best performance for

  * instantiation.  Everything is thread-safe.

  *

  * @param <I> Vertex id

  * @param <V> Vertex data

  * @param <E> Edge data

  */

 @SuppressWarnings("unchecked")

 public class ImmutableClassesGiraphConfiguration<I extends WritableComparable,

     V extends Writable, E extends Writable> extends GiraphConfiguration {

   /** Holder for all the classes */

   private final GiraphClasses classes;

   /** Mapping target class */

   private Class<? extends Writable> mappingTargetClass = null;

   /** Value (IVEMM) Factories */

   private final ValueFactories<I, V, E> valueFactories;

   /** Factory to create {@link OutEdges} for computation */

   private final OutEdgesFactory<I, E> outEdgesFactory;

   /** Factory to create {@link OutEdges} for input */

   private final OutEdgesFactory<I, E> inputOutEdgesFactory;

   /** Language values (IVEMM) are implemented in */

   private final PerGraphTypeEnum<Language> valueLanguages;

   /** Whether values (IVEMM) need Jython wrappers */

   private final PerGraphTypeBoolean valueNeedsWrappers;

   /**

    * Use unsafe serialization? Cached for fast access to instantiate the

    * extended data input/output classes

    */

   private final boolean useUnsafeSerialization;

   /**

    * Use BigDataIO for messages? Cached for fast access to instantiate the

    * extended data input/output classes for messages

    */

   private final boolean useBigDataIOForMessages;

   /** Is the graph static (meaning there is no mutation)? */

   private final boolean isStaticGraph;

   /** Whether or not to use message size encoding */

   private final boolean useMessageSizeEncoding;

   /**

    * Constructor.  Takes the configuration and then gets the classes out of

    * them for Giraph

    *

    * @param conf Configuration

    */

   public ImmutableClassesGiraphConfiguration(Configuration conf) {

     super(conf);

     classes = new GiraphClasses<I, V, E>(conf);

     useUnsafeSerialization = USE_UNSAFE_SERIALIZATION.get(this);

     useBigDataIOForMessages = USE_BIG_DATA_IO_FOR_MESSAGES.get(this);

     valueLanguages = PerGraphTypeEnum.readFromConf(

         GiraphConstants.GRAPH_TYPE_LANGUAGES, conf);

     valueNeedsWrappers = PerGraphTypeBoolean.readFromConf(

         GiraphConstants.GRAPH_TYPES_NEEDS_WRAPPERS, conf);

     isStaticGraph = GiraphConstants.STATIC_GRAPH.get(this);

     valueFactories = new ValueFactories<I, V, E>(this);

     outEdgesFactory = VERTEX_EDGES_FACTORY_CLASS.newInstance(this);

     inputOutEdgesFactory = INPUT_VERTEX_EDGES_FACTORY_CLASS.newInstance(this);

     useMessageSizeEncoding = USE_MESSAGE_SIZE_ENCODING.get(conf);

   }

   /**

    * Configure an object with this instance if the object is configurable.

    *

    * @param obj Object

    */

   public void configureIfPossible(Object obj) {

     if (obj instanceof GiraphConfigurationSettable) {

       ((GiraphConfigurationSettable) obj).setConf(this);

     }

   }

   public PerGraphTypeBoolean getValueNeedsWrappers() {

     return valueNeedsWrappers;

   }

   public PerGraphTypeEnum<Language> getValueLanguages() {

     return valueLanguages;

   }

   /**

    * Get the class used for edge translation during vertex input

    *

    * @return edge translation class

    */

   public Class<? extends TranslateEdge> edgeTranslationClass() {

     return EDGE_TRANSLATION_CLASS.get(this);

   }

   /**

    * Instance of TranslateEdge that contains helper method for edge translation

    *

    * @return instance of TranslateEdge

    */

   public TranslateEdge<I, E> edgeTranslationInstance() {

     if (edgeTranslationClass() != null) {

       return ReflectionUtils.newInstance(edgeTranslationClass(), this);

     }

     return null;

   }

   /**

    * Get the vertex input filter class

    *

    * @return VertexInputFilter class

    */

   public Class<? extends EdgeInputFilter<I, E>>

   getEdgeInputFilterClass() {

     return classes.getEdgeInputFilterClass();

   }

   /**

    * Get the edge input filter to use

    *

    * @return EdgeInputFilter

    */

   public EdgeInputFilter getEdgeInputFilter() {

     return ReflectionUtils.newInstance(getEdgeInputFilterClass(), this);

   }

   /**

    * Get the vertex input filter class

    *

    * @return VertexInputFilter class

    */

   public Class<? extends VertexInputFilter<I, V, E>>

   getVertexInputFilterClass() {

     return classes.getVertexInputFilterClass();

   }

   /**

    * Get the vertex input filter to use

    *

    * @return VertexInputFilter

    */

   public VertexInputFilter getVertexInputFilter() {

     return ReflectionUtils.newInstance(getVertexInputFilterClass(), this);

   }

   /**

    * Get the user's subclassed

    * {@link org.apache.giraph.partition.GraphPartitionerFactory}.

    *

    * @return User's graph partitioner

    */

   public Class<? extends GraphPartitionerFactory<I, V, E>>

   getGraphPartitionerClass() {

     return classes.getGraphPartitionerFactoryClass();

   }

   /**

    * Create a user graph partitioner class

    *

    * @return Instantiated user graph partitioner class

    */

   public GraphPartitionerFactory<I, V, E> createGraphPartitioner() {

     Class<? extends GraphPartitionerFactory<I, V, E>> klass =

         classes.getGraphPartitionerFactoryClass();

     return ReflectionUtils.newInstance(klass, this);

   }

   @Override

   public boolean hasVertexInputFormat() {

     return classes.hasVertexInputFormat();

   }

   /**

    * Get the user's subclassed

    * {@link org.apache.giraph.io.VertexInputFormat}.

    *

    * @return User's vertex input format class

    */

   public Class<? extends VertexInputFormat<I, V, E>>

   getVertexInputFormatClass() {

     return classes.getVertexInputFormatClass();

   }

   /**

    * Create a user vertex input format class.

    * Note: Giraph should only use WrappedVertexInputFormat,

    * which makes sure that Configuration parameters are set properly.

    *

    * @return Instantiated user vertex input format class

    */

   private VertexInputFormat<I, V, E> createVertexInputFormat() {

     Class<? extends VertexInputFormat<I, V, E>> klass =

         getVertexInputFormatClass();

     return ReflectionUtils.newInstance(klass, this);

   }

   /**

    * Create a wrapper for user vertex input format,

    * which makes sure that Configuration parameters are set properly in all

    * methods related to this format.

    *

    * @return Wrapper around user vertex input format

    */

   public WrappedVertexInputFormat<I, V, E> createWrappedVertexInputFormat() {

     WrappedVertexInputFormat<I, V, E> wrappedVertexInputFormat =

         new WrappedVertexInputFormat<I, V, E>(createVertexInputFormat());

     configureIfPossible(wrappedVertexInputFormat);

     return wrappedVertexInputFormat;

   }

   @Override

   public void setVertexInputFormatClass(

       Class<? extends VertexInputFormat> vertexInputFormatClass) {

     super.setVertexInputFormatClass(vertexInputFormatClass);

     classes.setVertexInputFormatClass(vertexInputFormatClass);

   }

   @Override

   public boolean hasVertexOutputFormat() {

     return classes.hasVertexOutputFormat();

   }

   /**

    * Get the user's subclassed

    * {@link org.apache.giraph.io.VertexOutputFormat}.

    *

    * @return User's vertex output format class

    */

   public Class<? extends VertexOutputFormat<I, V, E>>

   getVertexOutputFormatClass() {

     return classes.getVertexOutputFormatClass();

   }

   /**

    * Get MappingInputFormatClass

    *

    * @return MappingInputFormatClass

    */

   public Class<? extends MappingInputFormat<I, V, E, ? extends Writable>>

   getMappingInputFormatClass() {

     return classes.getMappingInputFormatClass();

   }

   /**

    * Set MappingInputFormatClass

    *

    * @param mappingInputFormatClass Determines how mappings are input

    */

   @Override

   public void setMappingInputFormatClass(

     Class<? extends MappingInputFormat> mappingInputFormatClass) {

     super.setMappingInputFormatClass(mappingInputFormatClass);

     classes.setMappingInputFormatClass(mappingInputFormatClass);

   }

   /**

    * Check if mappingInputFormat is set

    *

    * @return true if mappingInputFormat is set

    */

   public boolean hasMappingInputFormat() {

     return classes.hasMappingInputFormat();

   }

   /**

    * Create a user vertex output format class.

    * Note: Giraph should only use WrappedVertexOutputFormat,

    * which makes sure that Configuration parameters are set properly.

    *

    * @return Instantiated user vertex output format class

    */

   private VertexOutputFormat<I, V, E> createVertexOutputFormat() {

     Class<? extends VertexOutputFormat<I, V, E>> klass =

         getVertexOutputFormatClass();

     return ReflectionUtils.newInstance(klass, this);

   }

   /**

    * Create a user mapping input format class.

    * Note: Giraph should only use WrappedMappingInputFormat,

    * which makes sure that Configuration parameters are set properly.

    *

    * @return Instantiated user mapping input format class

    */

   private MappingInputFormat<I, V, E, ? extends Writable>

   createMappingInputFormat() {

     Class<? extends MappingInputFormat<I, V, E, ? extends Writable>> klass =

         getMappingInputFormatClass();

     return ReflectionUtils.newInstance(klass, this);

   }

   /**

    * Create a wrapper for user vertex output format,

    * which makes sure that Configuration parameters are set properly in all

    * methods related to this format.

    *

    * @return Wrapper around user vertex output format

    */

   public WrappedVertexOutputFormat<I, V, E> createWrappedVertexOutputFormat() {

     WrappedVertexOutputFormat<I, V, E> wrappedVertexOutputFormat =

         new WrappedVertexOutputFormat<I, V, E>(createVertexOutputFormat());

     configureIfPossible(wrappedVertexOutputFormat);

     return wrappedVertexOutputFormat;

   }

   /**

    * Create a wrapper for user mapping input format,

    * which makes sure that Configuration parameters are set properly in all

    * methods related to this format.

    *

    * @return Wrapper around user mapping input format

    */

   public WrappedMappingInputFormat<I, V, E, ? extends Writable>

   createWrappedMappingInputFormat() {

     WrappedMappingInputFormat<I, V, E, ? extends Writable>

       wrappedMappingInputFormat =

         new WrappedMappingInputFormat<>(createMappingInputFormat());

     configureIfPossible(wrappedMappingInputFormat);

     return wrappedMappingInputFormat;

   }

   @Override

   public boolean hasEdgeOutputFormat() {

     return classes.hasEdgeOutputFormat();

   }

   /**

    * Get the user's subclassed

    * {@link org.apache.giraph.io.EdgeOutputFormat}.

    *

    * @return User's edge output format class

    */

   public Class<? extends EdgeOutputFormat<I, V, E>>

   getEdgeOutputFormatClass() {

     return classes.getEdgeOutputFormatClass();

   }

   /**

    * Create a user edge output format class.

    * Note: Giraph should only use WrappedEdgeOutputFormat,

    * which makes sure that Configuration parameters are set properly.

    *

    * @return Instantiated user edge output format class

    */

   private EdgeOutputFormat<I, V, E> createEdgeOutputFormat() {

     Class<? extends EdgeOutputFormat<I, V, E>> klass =

         getEdgeOutputFormatClass();

     return ReflectionUtils.newInstance(klass, this);

   }

   /**

    * Create a wrapper for user edge output format,

    * which makes sure that Configuration parameters are set properly in all

    * methods related to this format.

    *

    * @return Wrapper around user edge output format

    */

   public WrappedEdgeOutputFormat<I, V, E> createWrappedEdgeOutputFormat() {

     WrappedEdgeOutputFormat<I, V, E> wrappedEdgeOutputFormat =

         new WrappedEdgeOutputFormat<I, V, E>(createEdgeOutputFormat());

     configureIfPossible(wrappedEdgeOutputFormat);

     return wrappedEdgeOutputFormat;

   }

   /**

    * Create the proper superstep output, based on the configuration settings.

    *

    * @param context Mapper context

    * @return SuperstepOutput

    */

   public SuperstepOutput<I, V, E> createSuperstepOutput(

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

     if (doOutputDuringComputation()) {

       if (vertexOutputFormatThreadSafe()) {

         return new MultiThreadedSuperstepOutput<I, V, E>(this, context);

       } else {

         return new SynchronizedSuperstepOutput<I, V, E>(this, context);

       }

     } else {

       return new NoOpSuperstepOutput<I, V, E>();

     }

   }

   @Override

   public boolean hasEdgeInputFormat() {

     return classes.hasEdgeInputFormat();

   }

   /**

    * Get the user's subclassed

    * {@link org.apache.giraph.io.EdgeInputFormat}.

    *

    * @return User's edge input format class

    */

   public Class<? extends EdgeInputFormat<I, E>> getEdgeInputFormatClass() {

     return classes.getEdgeInputFormatClass();

   }

   /**

    * Create a user edge input format class.

    * Note: Giraph should only use WrappedEdgeInputFormat,

    * which makes sure that Configuration parameters are set properly.

    *

    * @return Instantiated user edge input format class

    */

   private EdgeInputFormat<I, E> createEdgeInputFormat() {

     Class<? extends EdgeInputFormat<I, E>> klass = getEdgeInputFormatClass();

     return ReflectionUtils.newInstance(klass, this);

   }

   /**

    * Create a wrapper for user edge input format,

    * which makes sure that Configuration parameters are set properly in all

    * methods related to this format.

    *

    * @return Wrapper around user edge input format

    */

   public WrappedEdgeInputFormat<I, E> createWrappedEdgeInputFormat() {

     WrappedEdgeInputFormat<I, E> wrappedEdgeInputFormat =

         new WrappedEdgeInputFormat<I, E>(createEdgeInputFormat());

     configureIfPossible(wrappedEdgeInputFormat);

     return wrappedEdgeInputFormat;

   }

   @Override

   public void setEdgeInputFormatClass(

       Class<? extends EdgeInputFormat> edgeInputFormatClass) {

     super.setEdgeInputFormatClass(edgeInputFormatClass);

     classes.setEdgeInputFormatClass(edgeInputFormatClass);

   }

   /**

    * Get the user's subclassed {@link AggregatorWriter}.

    *

    * @return User's aggregator writer class

    */

   public Class<? extends AggregatorWriter> getAggregatorWriterClass() {

     return classes.getAggregatorWriterClass();

   }

   /**

    * Create a user aggregator output format class

    *

    * @return Instantiated user aggregator writer class

    */

   public AggregatorWriter createAggregatorWriter() {

     return ReflectionUtils.newInstance(getAggregatorWriterClass(), this);

   }

   /**

    * Get the user's subclassed

    * {@link org.apache.giraph.graph.VertexValueCombiner} class.

    *

    * @return User's vertex value combiner class

    */

   public Class<? extends VertexValueCombiner<V>>

   getVertexValueCombinerClass() {

     return classes.getVertexValueCombinerClass();

   }

   /**

    * Create a user vertex value combiner class

    *

    * @return Instantiated user vertex value combiner class

    */

   @SuppressWarnings("rawtypes")

   public VertexValueCombiner<V> createVertexValueCombiner() {

     return ReflectionUtils.newInstance(getVertexValueCombinerClass(), this);

   }

   /**

    * Get the user's subclassed VertexResolver.

    *

    * @return User's vertex resolver class

    */

   public Class<? extends VertexResolver<I, V, E>> getVertexResolverClass() {

     return classes.getVertexResolverClass();

   }

   /**

    * Create a user vertex revolver

    *

    * @return Instantiated user vertex resolver

    */

   public VertexResolver<I, V, E> createVertexResolver() {

     return ReflectionUtils.newInstance(getVertexResolverClass(), this);

   }

   /**

    * Get the user's subclassed WorkerContext.

    *

    * @return User's worker context class

    */

   public Class<? extends WorkerContext> getWorkerContextClass() {

     return classes.getWorkerContextClass();

   }

   /**

    * Create a user worker context

    *

    * @return Instantiated user worker context

    */

   public WorkerContext createWorkerContext() {

     return ReflectionUtils.newInstance(getWorkerContextClass(), this);

   }

   /**

    * Get the user's subclassed {@link org.apache.giraph.master.MasterCompute}

    *

    * @return User's master class

    */

   public Class<? extends MasterCompute> getMasterComputeClass() {

     return classes.getMasterComputeClass();

   }

   /**

    * Create a user master

    *

    * @return Instantiated user master

    */

   public MasterCompute createMasterCompute() {

     return ReflectionUtils.newInstance(getMasterComputeClass(), this);

   }

   @Override

   public Class<? extends Computation<I, V, E,

       ? extends Writable, ? extends Writable>>

   getComputationClass() {

     return classes.getComputationClass();

   }

   /**

    * Get computation factory class

    *

    * @return computation factory class

    */

   @Override

   public Class<? extends ComputationFactory<I, V, E,

       ? extends Writable, ? extends Writable>>

   getComputationFactoryClass() {

     return classes.getComputationFactoryClass();

   }

   /**

    * Get computation factory

    *

    * @return computation factory

    */

   public ComputationFactory<I, V, E, ? extends Writable, ? extends Writable>

   createComputationFactory() {

     return ReflectionUtils.newInstance(getComputationFactoryClass(), this);

   }

   /**

    * Create a user computation

    *

    * @return Instantiated user computation

    */

   public Computation<I, V, E, ? extends Writable, ? extends Writable>

   createComputation() {

     return createComputationFactory().createComputation(this);

   }

   /**

    * Get user types describing graph (I,V,E,M1,M2)

    *

    * @return GiraphTypes

    */

   public GiraphTypes<I, V, E> getGiraphTypes() {

     return classes.getGiraphTypes();

   }

   /**

    * Create a vertex

    *

    * @return Instantiated vertex

    */

   public Vertex<I, V, E> createVertex() {

     Class vertexClass = classes.getVertexClass();

     return (Vertex<I, V, E>) ReflectionUtils.newInstance(vertexClass, this);

   }

  /**

    * Get the user's subclassed vertex index class.

    *

    * @return User's vertex index class

    */

   public Class<I> getVertexIdClass() {

     return classes.getVertexIdClass();

   }

   /**

    * Get vertex ID factory

    *

    * @return {@link VertexIdFactory}

    */

   public VertexIdFactory<I> getVertexIdFactory() {

     return valueFactories.getVertexIdFactory();

   }

   /**

    * Create a user vertex index

    *

    * @return Instantiated user vertex index

    */

   public I createVertexId() {

     return getVertexIdFactory().newInstance();

   }

   /**

    * Get the user's subclassed vertex value class.

    *

    * @return User's vertex value class

    */

   public Class<V> getVertexValueClass() {

     return classes.getVertexValueClass();

   }

   /**

    * Get vertex value factory

    *

    * @return {@link VertexValueFactory}

    */

   public VertexValueFactory<V> getVertexValueFactory() {

     return valueFactories.getVertexValueFactory();

   }

   /**

    * Create a user vertex value

    *

    * @return Instantiated user vertex value

    */

   @SuppressWarnings("unchecked")

   public V createVertexValue() {

     return getVertexValueFactory().newInstance();

   }

   /**

    * Get the user's subclassed vertex value factory class

    *

    * @return User's vertex value factory class

    */

   public Class<? extends VertexValueFactory<V>> getVertexValueFactoryClass() {

     return (Class<? extends VertexValueFactory<V>>)

         valueFactories.getVertexValueFactory().getClass();

   }

   /**

    * Create array of MasterObservers.

    *

    * @param context Mapper context

    * @return Instantiated array of MasterObservers.

    */

   public MasterObserver[] createMasterObservers(

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

     Class<? extends MasterObserver>[] klasses = getMasterObserverClasses();

     MasterObserver[] objects = new MasterObserver[klasses.length];

     for (int i = 0; i < klasses.length; ++i) {

       objects[i] = ReflectionUtils.newInstance(klasses[i], this, context);

     }

     return objects;

   }

   /**

    * Create array of WorkerObservers.

    *

    * @param context Mapper context

    * @return Instantiated array of WorkerObservers.

    */

   public WorkerObserver[] createWorkerObservers(

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

     Class<? extends WorkerObserver>[] klasses = getWorkerObserverClasses();

     WorkerObserver[] objects = new WorkerObserver[klasses.length];

     for (int i = 0; i < klasses.length; ++i) {

       objects[i] = ReflectionUtils.newInstance(klasses[i], this, context);

     }

     return objects;

   }

   /**

    * Create array of MapperObservers.

    *

    * @param context Mapper context

    * @return Instantiated array of MapperObservers.

    */

   public MapperObserver[] createMapperObservers(

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

     Class<? extends MapperObserver>[] klasses = getMapperObserverClasses();

     MapperObserver[] objects = new MapperObserver[klasses.length];

     for (int i = 0; i < klasses.length; ++i) {

       objects[i] = ReflectionUtils.newInstance(klasses[i], this, context);

     }

     return objects;

   }

   /**

    * Create array of GcObservers.

    *

    * @param context Mapper context

    * @return Instantiated array of GcObservers.

    */

   public GcObserver[] createGcObservers(

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

     Class<? extends GcObserver>[] klasses = getGcObserverClasses();

     GcObserver[] objects = new GcObserver[klasses.length];

     for (int i = 0; i < klasses.length; ++i) {

       objects[i] = ReflectionUtils.newInstance(klasses[i], this, context);

     }

     return objects;

   }

   /**

    * Create job observer

    *

    * @return GiraphJobObserver set in configuration.

    */

   public GiraphJobObserver getJobObserver() {

     return ReflectionUtils.newInstance(getJobObserverClass(), this);

   }

   /**

    * Create job retry checker

    *

    * @return GiraphJobRetryChecker set in configuration.

    */

   public GiraphJobRetryChecker getJobRetryChecker() {

     return ReflectionUtils.newInstance(getJobRetryCheckerClass(), this);

   }

   /**

    * Get the user's subclassed edge value class.

    *

    * @return User's vertex edge value class

    */

   public Class<E> getEdgeValueClass() {

     return classes.getEdgeValueClass();

   }

   /**

    * Tell if we are using NullWritable for Edge value.

    *

    * @return true if NullWritable is class for

    */

   public boolean isEdgeValueNullWritable() {

     return getEdgeValueClass() == NullWritable.class;

   }

   /**

    * Get Factory for creating edge values

    *

    * @return {@link EdgeValueFactory}

    */

   public EdgeValueFactory<E> getEdgeValueFactory() {

     return valueFactories.getEdgeValueFactory();

   }

   /**

    * Create a user edge value

    *

    * @return Instantiated user edge value

    */

   public E createEdgeValue() {

     return getEdgeValueFactory().newInstance();

   }

   /**

    * Create a user edge.

    *

    * @return Instantiated user edge.

    */

   public Edge<I, E> createEdge() {

     if (isEdgeValueNullWritable()) {

       return (Edge<I, E>) EdgeFactory.create(createVertexId());

     } else {

       return EdgeFactory.create(createVertexId(), createEdgeValue());

     }

   }

   /**

    * Create edge based on #createEdge definition

    *

    * @param translateEdge instance of TranslateEdge

    * @param edge edge to be translated

    * @return translated edge

    */

   public Edge<I, E> createEdge(TranslateEdge<I, E>

     translateEdge, Edge<I, E> edge) {

     I translatedId = translateEdge.translateId(edge.getTargetVertexId());

     if (isEdgeValueNullWritable()) {

       return (Edge<I, E>) EdgeFactory.create(translatedId);

     } else {

       return EdgeFactory.create(translatedId,

         translateEdge.cloneValue(edge.getValue()));

     }

   }

   /**

    * Create a reusable edge.

    *

    * @return Instantiated reusable edge.

    */

   public ReusableEdge<I, E> createReusableEdge() {

     if (isEdgeValueNullWritable()) {

       return (ReusableEdge<I, E>) EdgeFactory.createReusable(createVertexId());

     } else {

       return EdgeFactory.createReusable(createVertexId(), createEdgeValue());

     }

   }

   /**

    * Create edge store factory

    *

    * @return edge store factory

    */

   public EdgeStoreFactory<I, V, E> createEdgeStoreFactory() {

     Class<? extends EdgeStoreFactory> edgeStoreFactoryClass =

         EDGE_STORE_FACTORY_CLASS.get(this);

     return ReflectionUtils.newInstance(edgeStoreFactoryClass);

   }

   /**

    * Get the user's subclassed incoming message value class.

    *

    * @param <M> Message data

    * @return User's vertex message value class

    */

   public <M extends Writable> Class<M> getIncomingMessageValueClass() {

     return classes.getIncomingMessageClasses().getMessageClass();

   }

   /**

    * Get the user's subclassed outgoing message value class.

    *

    * @param <M> Message type

    * @return User's vertex message value class

    */

   public <M extends Writable> Class<M> getOutgoingMessageValueClass() {

     return classes.getOutgoingMessageClasses().getMessageClass();

   }

   /**

    * Get incoming message classes

    * @param <M> message type

    * @return incoming message classes

    */

   public <M extends Writable>

   MessageClasses<I, M> getIncomingMessageClasses() {

     return classes.getIncomingMessageClasses();

   }

   /**

    * Get outgoing message classes

    * @param <M> message type

    * @return outgoing message classes

    */

   public <M extends Writable>

   MessageClasses<I, M> getOutgoingMessageClasses() {

     return classes.getOutgoingMessageClasses();

   }

   /**

    * Create new outgoing message value factory

    * @param <M> message type

    * @return outgoing message value factory

    */

   public <M extends Writable>

   MessageValueFactory<M> createOutgoingMessageValueFactory() {

     return classes.getOutgoingMessageClasses().createMessageValueFactory(this);

   }

   /**

    * Create new incoming message value factory

    * @param <M> message type

    * @return incoming message value factory

    */

   public <M extends Writable>

   MessageValueFactory<M> createIncomingMessageValueFactory() {

     return classes.getIncomingMessageClasses().createMessageValueFactory(this);

   }

   @Override

   public void setMessageCombinerClass(

       Class<? extends MessageCombiner> messageCombinerClass) {

     throw new IllegalArgumentException(

         "Cannot set message combiner on ImmutableClassesGiraphConfigurable");

   }

   /**

    * Create a user combiner class

    *

    * @param <M> Message data

    * @return Instantiated user combiner class

    */

   public <M extends Writable> MessageCombiner<? super I, M>

   createOutgoingMessageCombiner() {

     return classes.getOutgoingMessageClasses().createMessageCombiner(this);

   }

   /**

    * Check if user set a combiner

    *

    * @return True iff user set a combiner class

    */

   public boolean useOutgoingMessageCombiner() {

     return classes.getOutgoingMessageClasses().useMessageCombiner();

   }

   /**

    * Get outgoing message encode and store type

    * @return outgoing message encode and store type

    */

   public MessageEncodeAndStoreType getOutgoingMessageEncodeAndStoreType() {

     return classes.getOutgoingMessageClasses().getMessageEncodeAndStoreType();

   }

   @Override

   public Class<? extends OutEdges<I, E>> getOutEdgesClass() {

     return classes.getOutEdgesClass();

   }

   /**

    * Get the user's subclassed {@link org.apache.giraph.edge.OutEdges} used for

    * input

    *

    * @return User's input vertex edges class

    */

   public Class<? extends OutEdges<I, E>> getInputOutEdgesClass() {

     return classes.getInputOutEdgesClass();

   }

   /**

    * Check whether the user has specified a different

    * {@link org.apache.giraph.edge.OutEdges} class to be used during

    * edge-based input.

    *

    * @return True iff there is a special edges class for input

    */

   public boolean useInputOutEdges() {

     return classes.getInputOutEdgesClass() != classes.getOutEdgesClass();

   }

   /**

    * Get MappingStore class to be used

    *

    * @return MappingStore class set by user

    */

   public Class<? extends MappingStore> getMappingStoreClass() {

     return MAPPING_STORE_CLASS.get(this);

   }

   /**

    * Create a {@link org.apache.giraph.mapping.MappingStore} instance

    *

    * @return MappingStore Instance

    */

   public MappingStore<I, ? extends Writable> createMappingStore() {

     if (getMappingStoreClass() != null) {

       return ReflectionUtils.newInstance(getMappingStoreClass(), this);

     } else {

       return null;

     }

   }

   /**

    * Get MappingStoreOps class to be used

    *

    * @return MappingStoreOps class set by user

    */

   public Class<? extends MappingStoreOps> getMappingStoreOpsClass() {

     return MAPPING_STORE_OPS_CLASS.get(this);

   }

   /**

    * Create a {@link org.apache.giraph.mapping.MappingStoreOps} instance

    *

    * @return MappingStoreOps Instance

    */

   public MappingStoreOps<I, ? extends Writable> createMappingStoreOps() {

     if (getMappingStoreOpsClass() != null) {

       return ReflectionUtils.newInstance(getMappingStoreOpsClass(), this);

     } else {

       return null;

     }

   }

   /**

    * Get mappingTarget class

    *

    * @return mappingTarget class

    */

   public Class<? extends Writable> getMappingTargetClass() {

     if (mappingTargetClass == null) {

       Class<?>[] classList = ReflectionUtils.getTypeArguments(

         MappingStore.class, getMappingStoreClass());

       Preconditions.checkArgument(classList.length == 2);

       mappingTargetClass = (Class<? extends Writable>) classList[1];

     }

     return mappingTargetClass;

   }

   /**

    * Create and return mappingTarget instance

    *

    * @return mappingTarget instance

    */

   public Writable createMappingTarget() {

     return WritableUtils.createWritable(getMappingTargetClass());

   }

   /**

    * Create a user {@link org.apache.giraph.edge.OutEdges}

    *

    * @return Instantiated user OutEdges

    */

   public OutEdges<I, E> createOutEdges() {

     return outEdgesFactory.newInstance();

   }

   /**

    * Create a {@link org.apache.giraph.edge.OutEdges} instance and initialize

    * it with the default capacity.

    *

    * @return Instantiated OutEdges

    */

   public OutEdges<I, E> createAndInitializeOutEdges() {

     OutEdges<I, E> outEdges = createOutEdges();

     outEdges.initialize();

     return outEdges;

   }

   /**

    * Create a {@link org.apache.giraph.edge.OutEdges} instance and initialize

    * it with the given capacity (the number of edges that will be added).

    *

    * @param capacity Number of edges that will be added

    * @return Instantiated OutEdges

    */

   public OutEdges<I, E> createAndInitializeOutEdges(int capacity) {

     OutEdges<I, E> outEdges = createOutEdges();

     outEdges.initialize(capacity);

     return outEdges;

   }

   /**

    * Create a {@link org.apache.giraph.edge.OutEdges} instance and initialize

    * it with the given iterable of edges.

    *

    * @param edges Iterable of edges to add

    * @return Instantiated OutEdges

    */

   public OutEdges<I, E> createAndInitializeOutEdges(

       Iterable<Edge<I, E>> edges) {

     OutEdges<I, E> outEdges = createOutEdges();

     outEdges.initialize(edges);

     return outEdges;

   }

   /**

    * Create a user {@link org.apache.giraph.edge.OutEdges} used during

    * edge-based input

    *

    * @return Instantiated user input OutEdges

    */

   public OutEdges<I, E> createInputOutEdges() {

     return inputOutEdgesFactory.newInstance();

   }

   /**

    * Create an input {@link org.apache.giraph.edge.OutEdges} instance and

    * initialize it with the default capacity.

    *

    * @return Instantiated input OutEdges

    */

   public OutEdges<I, E> createAndInitializeInputOutEdges() {

     OutEdges<I, E> outEdges = createInputOutEdges();

     outEdges.initialize();

     return outEdges;

   }

   /**

    * Create a partition

    *

    * @param id Partition id

    * @param progressable Progressable for reporting progress

    * @return Instantiated partition

    */

   public Partition<I, V, E> createPartition(

       int id, Progressable progressable) {

     Class<? extends Partition<I, V, E>> klass = classes.getPartitionClass();

     Partition<I, V, E> partition = ReflectionUtils.newInstance(klass, this);

     partition.initialize(id, progressable);

     return partition;

   }

   /**

    * Use unsafe serialization?

    *

    * @return True if using unsafe serialization, false otherwise.

    */

   public boolean useUnsafeSerialization() {

     return useUnsafeSerialization;

   }

   /**

    * Create DataInputOutput to store messages

    *

    * @return DataInputOutput object

    */

   public DataInputOutput createMessagesInputOutput() {

     if (useBigDataIOForMessages) {

       return new BigDataInputOutput(this);

     } else {

       return new ExtendedDataInputOutput(this);

     }

   }

   /**

    * Create an extended data output (can be subclassed)

    *

    * @return ExtendedDataOutput object

    */

   public ExtendedDataOutput createExtendedDataOutput() {

     if (useUnsafeSerialization) {

       return new UnsafeByteArrayOutputStream();

     } else {

       return new ExtendedByteArrayDataOutput();

     }

   }

   /**

    * Create an extended data output (can be subclassed)

    *

    * @param expectedSize Expected size

    * @return ExtendedDataOutput object

    */

   public ExtendedDataOutput createExtendedDataOutput(int expectedSize) {

     if (useUnsafeSerialization) {

       return new UnsafeByteArrayOutputStream(expectedSize);

     } else {

       return new ExtendedByteArrayDataOutput(expectedSize);

     }

   }

   /**

    * Create an extended data output (can be subclassed)

    *

    * @param buf Buffer to use for the output (reuse perhaps)

    * @param pos How much of the buffer is already used

    * @return ExtendedDataOutput object

    */

   public ExtendedDataOutput createExtendedDataOutput(byte[] buf,

                                                      int pos) {

     if (useUnsafeSerialization) {

       return new UnsafeByteArrayOutputStream(buf, pos);

     } else {

       return new ExtendedByteArrayDataOutput(buf, pos);

     }

   }

   /**

    * Create an extended data input (can be subclassed)

    *

    * @param buf Buffer to use for the input

    * @param off Where to start reading in the buffer

    * @param length Maximum length of the buffer

    * @return ExtendedDataInput object

    */

   public ExtendedDataInput createExtendedDataInput(

       byte[] buf, int off, int length) {

     if (useUnsafeSerialization) {

       return new UnsafeByteArrayInputStream(buf, off, length);

     } else {

       return new ExtendedByteArrayDataInput(buf, off, length);

     }

   }

   /**

    * Create an extended data input (can be subclassed)

    *

    * @param buf Buffer to use for the input

    * @return ExtendedDataInput object

    */

   public ExtendedDataInput createExtendedDataInput(byte[] buf) {

     if (useUnsafeSerialization) {

       return new UnsafeByteArrayInputStream(buf);

     } else {

       return new ExtendedByteArrayDataInput(buf);

     }

   }

   /**

    * Create extendedDataInput based on extendedDataOutput

    *

    * @param extendedDataOutput extendedDataOutput

    * @return extendedDataInput

    */

   public ExtendedDataInput createExtendedDataInput(

     ExtendedDataOutput extendedDataOutput) {

     return createExtendedDataInput(extendedDataOutput.getByteArray(), 0,

         extendedDataOutput.getPos());

   }

   /**

    * Whether to use an unsafe serialization

    *

    * @return whether to use unsafe serialization

    */

   public boolean getUseUnsafeSerialization() {

     return useUnsafeSerialization;

   }

   /**

    * Update Computation and MessageCombiner class used

    *

    * @param superstepClasses SuperstepClasses