giraph-core/src/main/java/org/apache/giraph/job/GiraphConfigurationValidator.java - giraph - Git at Google

 /*
  * 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.job;

 import static org.apache.giraph.conf.GiraphConstants.VERTEX_EDGES_CLASS;
 import static org.apache.giraph.conf.GiraphConstants.VERTEX_RESOLVER_CLASS;
 import static org.apache.giraph.utils.ReflectionUtils.getTypeArguments;

 import org.apache.giraph.combiner.MessageCombiner;
 import org.apache.giraph.conf.GiraphConstants;
 import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
 import org.apache.giraph.edge.OutEdges;
 import org.apache.giraph.factories.DefaultVertexValueFactory;
 import org.apache.giraph.factories.VertexValueFactory;
 import org.apache.giraph.graph.DefaultVertexResolver;
 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.VertexInputFormat;
 import org.apache.giraph.io.VertexOutputFormat;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableComparable;
 import org.apache.log4j.Logger;

 /**
  * GiraphConfigurationValidator attempts to verify the consistency of
  * user-chosen InputFormat, OutputFormat, and Vertex generic type
  * parameters as well as the general Configuration settings
  * before the job run actually begins.
  *
  * @param <I> the Vertex ID type
  * @param <V> the Vertex Value type
  * @param <E> the Edge Value type
  * @param <M1> the incoming Message type
  * @param <M2> the outgoing Message type
  */
 public class GiraphConfigurationValidator<I extends WritableComparable,
     V extends Writable, E extends Writable, M1 extends Writable,
     M2 extends Writable> {
   /**
    * Class logger object.
    */
   private static Logger LOG =
     Logger.getLogger(GiraphConfigurationValidator.class);

   /** I param vertex index in classList */
   private static final int ID_PARAM_INDEX = 0;
   /** V param vertex index in classList */
   private static final int VALUE_PARAM_INDEX = 1;
   /** E param vertex index in classList */
   private static final int EDGE_PARAM_INDEX = 2;
   /** M param vertex combiner index in classList */
   private static final int MSG_COMBINER_PARAM_INDEX = 1;
   /** E param edge input format index in classList */
   private static final int EDGE_PARAM_EDGE_INPUT_FORMAT_INDEX = 1;
   /** E param vertex edges index in classList */
   private static final int EDGE_PARAM_OUT_EDGES_INDEX = 1;
   /** V param vertex value factory index in classList */
   private static final int VALUE_PARAM_VERTEX_VALUE_FACTORY_INDEX = 0;
   /** V param vertex value combiner index in classList */
   private static final int VALUE_PARAM_VERTEX_VALUE_COMBINER_INDEX = 0;

   /**
    * The Configuration object for use in the validation test.
    */
   private final ImmutableClassesGiraphConfiguration conf;

   /**
    * Constructor to execute the validation test, throws
    * unchecked exception to end job run on failure.
    *
    * @param conf the Configuration for this run.
    */
   public GiraphConfigurationValidator(Configuration conf) {
     this.conf = new ImmutableClassesGiraphConfiguration(conf);
   }

   /**
    * Get vertex id type
    *
    * @return vertex id type
    */
   private Class<? extends WritableComparable> vertexIndexType() {
     return conf.getGiraphTypes().getVertexIdClass();
   }

   /**
    * Get vertex value type
    *
    * @return vertex value type
    */
   private Class<? extends Writable> vertexValueType() {
     return conf.getGiraphTypes().getVertexValueClass();
   }

   /**
    * Get edge value type
    *
    * @return edge value type
    */
   private Class<? extends Writable> edgeValueType() {
     return conf.getGiraphTypes().getEdgeValueClass();
   }

   /**
    * Get outgoing message value type
    *
    * @return outgoing message value type
    */
   private Class<? extends Writable> outgoingMessageValueType() {
     return conf.getOutgoingMessageValueClass();
   }

   /**
    * Make sure that all registered classes have matching types.  This
    * is a little tricky due to type erasure, cannot simply get them from
    * the class type arguments.  Also, set the vertex index, vertex value,
    * edge value and message value classes.
    */
   public void validateConfiguration() {
     checkConfiguration();
     verifyOutEdgesGenericTypes();
     verifyVertexInputFormatGenericTypes();
     verifyEdgeInputFormatGenericTypes();
     verifyVertexOutputFormatGenericTypes();
     verifyEdgeOutputFormatGenericTypes();
     verifyVertexResolverGenericTypes();
     verifyVertexValueCombinerGenericTypes();
     verifyMessageCombinerGenericTypes();
     verifyVertexValueFactoryGenericTypes();
   }

   /**
    * Make sure the configuration is set properly by the user prior to
    * submitting the job.
    */
   private void checkConfiguration() {
     if (conf.getMaxWorkers() < 0) {
       throw new RuntimeException("checkConfiguration: No valid " +
           GiraphConstants.MAX_WORKERS);
     }
     if (conf.getMinPercentResponded() <= 0.0f ||
         conf.getMinPercentResponded() > 100.0f) {
       throw new IllegalArgumentException(
           "checkConfiguration: Invalid " + conf.getMinPercentResponded() +
               " for " + GiraphConstants.MIN_PERCENT_RESPONDED.getKey());
     }
     if (conf.getMinWorkers() < 0) {
       throw new IllegalArgumentException("checkConfiguration: No valid " +
           GiraphConstants.MIN_WORKERS);
     }
     conf.createComputationFactory().checkConfiguration(conf);
     if (conf.getVertexInputFormatClass() == null &&
         conf.getEdgeInputFormatClass() == null) {
       throw new IllegalArgumentException("checkConfiguration: One of " +
           GiraphConstants.VERTEX_INPUT_FORMAT_CLASS.getKey() + " and " +
           GiraphConstants.EDGE_INPUT_FORMAT_CLASS.getKey() +
           " must be non-null");
     }
     if (conf.getVertexResolverClass() == null) {
       if (LOG.isInfoEnabled()) {
         LOG.info("checkConfiguration: No class found for " +
             VERTEX_RESOLVER_CLASS.getKey() +
             ", defaulting to " +
             VERTEX_RESOLVER_CLASS.getDefaultClass().getCanonicalName());
       }
     }
     if (conf.getOutEdgesClass() == null) {
       if (LOG.isInfoEnabled()) {
         LOG.info("checkConfiguration: No class found for " +
             VERTEX_EDGES_CLASS.getKey() + ", defaulting to " +
             VERTEX_EDGES_CLASS.getDefaultClass().getCanonicalName());
       }
     }
   }

   /**
    * Verify matching generic types for a specific OutEdges class.
    *
    * @param outEdgesClass {@link org.apache.giraph.edge.OutEdges} class to check
    */
   private void verifyOutEdgesGenericTypesClass(
       Class<? extends OutEdges<I, E>> outEdgesClass) {
     Class<?>[] classList = getTypeArguments(OutEdges.class, outEdgesClass);
     checkAssignable(classList, ID_PARAM_INDEX, vertexIndexType(),
         OutEdges.class, "vertex index");
     checkAssignable(classList, EDGE_PARAM_OUT_EDGES_INDEX, edgeValueType(),
         OutEdges.class, "edge value");
   }

   /** Verify matching generic types in OutEdges. */
   private void verifyOutEdgesGenericTypes() {
     Class<? extends OutEdges<I, E>> outEdgesClass =
         conf.getOutEdgesClass();
     Class<? extends OutEdges<I, E>> inputOutEdgesClass =
         conf.getInputOutEdgesClass();
     verifyOutEdgesGenericTypesClass(outEdgesClass);
     verifyOutEdgesGenericTypesClass(inputOutEdgesClass);
   }

   /** Verify matching generic types in VertexInputFormat. */
   private void verifyVertexInputFormatGenericTypes() {
     Class<? extends VertexInputFormat<I, V, E>> vertexInputFormatClass =
       conf.getVertexInputFormatClass();
     if (vertexInputFormatClass != null) {
       Class<?>[] classList =
           getTypeArguments(VertexInputFormat.class, vertexInputFormatClass);
       checkAssignable(classList, ID_PARAM_INDEX, vertexIndexType(),
           VertexInputFormat.class, "vertex index");
       checkAssignable(classList, VALUE_PARAM_INDEX, vertexValueType(),
           VertexInputFormat.class, "vertex value");
       checkAssignable(classList, EDGE_PARAM_INDEX, edgeValueType(),
           VertexInputFormat.class, "edge value");
     }
   }

   /** Verify matching generic types in EdgeInputFormat. */
   private void verifyEdgeInputFormatGenericTypes() {
     Class<? extends EdgeInputFormat<I, E>> edgeInputFormatClass =
         conf.getEdgeInputFormatClass();
     if (edgeInputFormatClass != null) {
       Class<?>[] classList =
           getTypeArguments(EdgeInputFormat.class, edgeInputFormatClass);
       checkAssignable(classList, ID_PARAM_INDEX, vertexIndexType(),
           EdgeInputFormat.class, "vertex index");
       checkAssignable(classList, EDGE_PARAM_EDGE_INPUT_FORMAT_INDEX,
           edgeValueType(), EdgeInputFormat.class, "edge value");
     }
   }

   /**
    * If there is a vertex value combiner type, verify its
    * generic params match the job.
    */
   private void verifyVertexValueCombinerGenericTypes() {
     Class<? extends VertexValueCombiner<V>> vertexValueCombiner =
         conf.getVertexValueCombinerClass();
     if (vertexValueCombiner != null) {
       Class<?>[] classList =
           getTypeArguments(VertexValueCombiner.class, vertexValueCombiner);
       checkAssignable(classList, VALUE_PARAM_VERTEX_VALUE_COMBINER_INDEX,
           vertexValueType(), VertexValueCombiner.class, "vertex value");
     }
   }

   /**
    * If there is a message combiner type, verify its
    * generic params match the job.
    */
   private void verifyMessageCombinerGenericTypes() {
     MessageCombiner<I, M2> messageCombiner =
       conf.createOutgoingMessageCombiner();
     if (messageCombiner != null) {
       Class<?>[] classList =
           getTypeArguments(MessageCombiner.class, messageCombiner.getClass());
       checkEquals(classList, ID_PARAM_INDEX, vertexIndexType(),
           MessageCombiner.class, "vertex index");
       checkEquals(classList, MSG_COMBINER_PARAM_INDEX,
           outgoingMessageValueType(), MessageCombiner.class, "message value");
     }
   }

   /** Verify that the vertex output format's generic params match the job. */
   private void verifyVertexOutputFormatGenericTypes() {
     Class<? extends EdgeOutputFormat<I, V, E>>
       edgeOutputFormatClass = conf.getEdgeOutputFormatClass();
     if (conf.hasEdgeOutputFormat()) {
       Class<?>[] classList =
         getTypeArguments(EdgeOutputFormat.class, edgeOutputFormatClass);
       checkAssignable(classList, ID_PARAM_INDEX, vertexIndexType(),
           VertexOutputFormat.class, "vertex index");
       checkAssignable(classList, VALUE_PARAM_INDEX, vertexValueType(),
           VertexOutputFormat.class, "vertex value");
       checkAssignable(classList, EDGE_PARAM_INDEX, edgeValueType(),
           VertexOutputFormat.class, "edge value");
     }
   }

   /** Verify that the edge output format's generic params match the job. */
   private void verifyEdgeOutputFormatGenericTypes() {
     Class<? extends VertexOutputFormat<I, V, E>>
       vertexOutputFormatClass = conf.getVertexOutputFormatClass();
     if (conf.hasVertexOutputFormat()) {
       Class<?>[] classList =
         getTypeArguments(VertexOutputFormat.class, vertexOutputFormatClass);
       checkAssignable(classList, ID_PARAM_INDEX, vertexIndexType(),
           VertexOutputFormat.class, "vertex index");
       checkAssignable(classList, VALUE_PARAM_INDEX, vertexValueType(),
           VertexOutputFormat.class, "vertex value");
       checkAssignable(classList, EDGE_PARAM_INDEX, edgeValueType(),
           VertexOutputFormat.class, "edge value");
     }
   }

   /** Verify that the vertex value factory's type matches the job */
   private void verifyVertexValueFactoryGenericTypes() {
     Class<? extends VertexValueFactory<V>>
         vvfClass = conf.getVertexValueFactoryClass();
     if (DefaultVertexValueFactory.class.equals(vvfClass)) {
       return;
     }
     Class<?>[] classList = getTypeArguments(VertexValueFactory.class, vvfClass);
     checkEquals(classList, VALUE_PARAM_VERTEX_VALUE_FACTORY_INDEX,
         vertexValueType(), VertexValueFactory.class, "vertex value");
   }

   /**
    * If there is a vertex resolver,
    * validate the generic parameter types.
    * */
   private void verifyVertexResolverGenericTypes() {
     Class<? extends VertexResolver<I, V, E>>
         vrClass = conf.getVertexResolverClass();
     if (DefaultVertexResolver.class.equals(vrClass)) {
       return;
     }
     Class<?>[] classList =
         getTypeArguments(VertexResolver.class, vrClass);
     checkEquals(classList, ID_PARAM_INDEX, vertexIndexType(),
         VertexResolver.class, "vertex index");
     checkEquals(classList, VALUE_PARAM_INDEX, vertexValueType(),
         VertexResolver.class, "vertex value");
     checkEquals(classList, EDGE_PARAM_INDEX, edgeValueType(),
         VertexResolver.class, "edge value");
   }

   /**
    * Check that the type from computation equals the type from the class.
    *
    * @param classList classes from type
    * @param index array index of class to check
    * @param classFromComputation class from computation
    * @param klass Class type we're checking, only used for printing name
    * @param typeName Name of type we're checking
    */
   private static void checkEquals(Class<?>[] classList, int index,
       Class<?> classFromComputation, Class klass, String typeName) {
     if (classList[index] == null) {
       LOG.warn(klass.getSimpleName() + " " + typeName + " type is not known");
     } else if (!classList[index].equals(classFromComputation)) {
       throw new IllegalStateException(
           "checkClassTypes: " + typeName + " types not equal, " +
               "computation - " + classFromComputation +
               ", " + klass.getSimpleName() + " - " +
               classList[index]);
     }
   }

   /**
    * Check that the type from computation is assignable to type from the class.
    *
    * @param classList classes from type
    * @param index array index of class to check
    * @param classFromComputation class from computation
    * @param klass Class type we're checking, only used for printing name
    * @param typeName Name of type we're checking
    */
   private static void checkAssignable(Class<?>[] classList, int index,
       Class<?> classFromComputation, Class klass, String typeName) {
     if (classList[index] == null) {
       LOG.warn(klass.getSimpleName() + " " + typeName + " type is not known");
     } else if (!classList[index].isAssignableFrom(classFromComputation)) {
       throw new IllegalStateException(
           "checkClassTypes: " + typeName + " types not assignable, " +
               "computation - " + classFromComputation +
               ", " + klass.getSimpleName() + " - " +
               classList[EDGE_PARAM_EDGE_INPUT_FORMAT_INDEX]);
     }
   }
 }
	/*
	* 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.job;

	import static org.apache.giraph.conf.GiraphConstants.VERTEX_EDGES_CLASS;
	import static org.apache.giraph.conf.GiraphConstants.VERTEX_RESOLVER_CLASS;
	import static org.apache.giraph.utils.ReflectionUtils.getTypeArguments;

	import org.apache.giraph.combiner.MessageCombiner;
	import org.apache.giraph.conf.GiraphConstants;
	import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
	import org.apache.giraph.edge.OutEdges;
	import org.apache.giraph.factories.DefaultVertexValueFactory;
	import org.apache.giraph.factories.VertexValueFactory;
	import org.apache.giraph.graph.DefaultVertexResolver;
	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.VertexInputFormat;
	import org.apache.giraph.io.VertexOutputFormat;
	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.io.Writable;
	import org.apache.hadoop.io.WritableComparable;
	import org.apache.log4j.Logger;

	/**
	* GiraphConfigurationValidator attempts to verify the consistency of
	* user-chosen InputFormat, OutputFormat, and Vertex generic type
	* parameters as well as the general Configuration settings
	* before the job run actually begins.
	*
	* @param <I> the Vertex ID type
	* @param <V> the Vertex Value type
	* @param <E> the Edge Value type
	* @param <M1> the incoming Message type
	* @param <M2> the outgoing Message type
	*/
	public class GiraphConfigurationValidator<I extends WritableComparable,
	V extends Writable, E extends Writable, M1 extends Writable,
	M2 extends Writable> {
	/**
	* Class logger object.
	*/
	private static Logger LOG =
	Logger.getLogger(GiraphConfigurationValidator.class);

	/** I param vertex index in classList */
	private static final int ID_PARAM_INDEX = 0;
	/** V param vertex index in classList */
	private static final int VALUE_PARAM_INDEX = 1;
	/** E param vertex index in classList */
	private static final int EDGE_PARAM_INDEX = 2;
	/** M param vertex combiner index in classList */
	private static final int MSG_COMBINER_PARAM_INDEX = 1;
	/** E param edge input format index in classList */
	private static final int EDGE_PARAM_EDGE_INPUT_FORMAT_INDEX = 1;
	/** E param vertex edges index in classList */
	private static final int EDGE_PARAM_OUT_EDGES_INDEX = 1;
	/** V param vertex value factory index in classList */
	private static final int VALUE_PARAM_VERTEX_VALUE_FACTORY_INDEX = 0;
	/** V param vertex value combiner index in classList */
	private static final int VALUE_PARAM_VERTEX_VALUE_COMBINER_INDEX = 0;

	/**
	* The Configuration object for use in the validation test.
	*/
	private final ImmutableClassesGiraphConfiguration conf;

	/**
	* Constructor to execute the validation test, throws
	* unchecked exception to end job run on failure.
	*
	* @param conf the Configuration for this run.
	*/
	public GiraphConfigurationValidator(Configuration conf) {
	this.conf = new ImmutableClassesGiraphConfiguration(conf);
	}

	/**
	* Get vertex id type
	*
	* @return vertex id type
	*/
	private Class<? extends WritableComparable> vertexIndexType() {
	return conf.getGiraphTypes().getVertexIdClass();
	}

	/**
	* Get vertex value type
	*
	* @return vertex value type
	*/
	private Class<? extends Writable> vertexValueType() {
	return conf.getGiraphTypes().getVertexValueClass();
	}

	/**
	* Get edge value type
	*
	* @return edge value type
	*/
	private Class<? extends Writable> edgeValueType() {
	return conf.getGiraphTypes().getEdgeValueClass();
	}

	/**
	* Get outgoing message value type
	*
	* @return outgoing message value type
	*/
	private Class<? extends Writable> outgoingMessageValueType() {
	return conf.getOutgoingMessageValueClass();
	}

	/**
	* Make sure that all registered classes have matching types. This
	* is a little tricky due to type erasure, cannot simply get them from
	* the class type arguments. Also, set the vertex index, vertex value,
	* edge value and message value classes.
	*/
	public void validateConfiguration() {
	checkConfiguration();
	verifyOutEdgesGenericTypes();
	verifyVertexInputFormatGenericTypes();
	verifyEdgeInputFormatGenericTypes();
	verifyVertexOutputFormatGenericTypes();
	verifyEdgeOutputFormatGenericTypes();
	verifyVertexResolverGenericTypes();
	verifyVertexValueCombinerGenericTypes();
	verifyMessageCombinerGenericTypes();
	verifyVertexValueFactoryGenericTypes();
	}

	/**
	* Make sure the configuration is set properly by the user prior to
	* submitting the job.
	*/
	private void checkConfiguration() {
	if (conf.getMaxWorkers() < 0) {
	throw new RuntimeException("checkConfiguration: No valid " +
	GiraphConstants.MAX_WORKERS);
	}
	if (conf.getMinPercentResponded() <= 0.0f \|\|
	conf.getMinPercentResponded() > 100.0f) {
	throw new IllegalArgumentException(
	"checkConfiguration: Invalid " + conf.getMinPercentResponded() +
	" for " + GiraphConstants.MIN_PERCENT_RESPONDED.getKey());
	}
	if (conf.getMinWorkers() < 0) {
	throw new IllegalArgumentException("checkConfiguration: No valid " +
	GiraphConstants.MIN_WORKERS);
	}
	conf.createComputationFactory().checkConfiguration(conf);
	if (conf.getVertexInputFormatClass() == null &&
	conf.getEdgeInputFormatClass() == null) {
	throw new IllegalArgumentException("checkConfiguration: One of " +
	GiraphConstants.VERTEX_INPUT_FORMAT_CLASS.getKey() + " and " +
	GiraphConstants.EDGE_INPUT_FORMAT_CLASS.getKey() +
	" must be non-null");
	}
	if (conf.getVertexResolverClass() == null) {
	if (LOG.isInfoEnabled()) {
	LOG.info("checkConfiguration: No class found for " +
	VERTEX_RESOLVER_CLASS.getKey() +
	", defaulting to " +
	VERTEX_RESOLVER_CLASS.getDefaultClass().getCanonicalName());
	}
	}
	if (conf.getOutEdgesClass() == null) {
	if (LOG.isInfoEnabled()) {
	LOG.info("checkConfiguration: No class found for " +
	VERTEX_EDGES_CLASS.getKey() + ", defaulting to " +
	VERTEX_EDGES_CLASS.getDefaultClass().getCanonicalName());
	}
	}
	}

	/**
	* Verify matching generic types for a specific OutEdges class.
	*
	* @param outEdgesClass {@link org.apache.giraph.edge.OutEdges} class to check
	*/
	private void verifyOutEdgesGenericTypesClass(
	Class<? extends OutEdges<I, E>> outEdgesClass) {
	Class<?>[] classList = getTypeArguments(OutEdges.class, outEdgesClass);
	checkAssignable(classList, ID_PARAM_INDEX, vertexIndexType(),
	OutEdges.class, "vertex index");
	checkAssignable(classList, EDGE_PARAM_OUT_EDGES_INDEX, edgeValueType(),
	OutEdges.class, "edge value");
	}

	/** Verify matching generic types in OutEdges. */
	private void verifyOutEdgesGenericTypes() {
	Class<? extends OutEdges<I, E>> outEdgesClass =
	conf.getOutEdgesClass();
	Class<? extends OutEdges<I, E>> inputOutEdgesClass =
	conf.getInputOutEdgesClass();
	verifyOutEdgesGenericTypesClass(outEdgesClass);
	verifyOutEdgesGenericTypesClass(inputOutEdgesClass);
	}

	/** Verify matching generic types in VertexInputFormat. */
	private void verifyVertexInputFormatGenericTypes() {
	Class<? extends VertexInputFormat<I, V, E>> vertexInputFormatClass =
	conf.getVertexInputFormatClass();
	if (vertexInputFormatClass != null) {
	Class<?>[] classList =
	getTypeArguments(VertexInputFormat.class, vertexInputFormatClass);
	checkAssignable(classList, ID_PARAM_INDEX, vertexIndexType(),
	VertexInputFormat.class, "vertex index");
	checkAssignable(classList, VALUE_PARAM_INDEX, vertexValueType(),
	VertexInputFormat.class, "vertex value");
	checkAssignable(classList, EDGE_PARAM_INDEX, edgeValueType(),
	VertexInputFormat.class, "edge value");
	}
	}

	/** Verify matching generic types in EdgeInputFormat. */
	private void verifyEdgeInputFormatGenericTypes() {
	Class<? extends EdgeInputFormat<I, E>> edgeInputFormatClass =
	conf.getEdgeInputFormatClass();
	if (edgeInputFormatClass != null) {
	Class<?>[] classList =
	getTypeArguments(EdgeInputFormat.class, edgeInputFormatClass);
	checkAssignable(classList, ID_PARAM_INDEX, vertexIndexType(),
	EdgeInputFormat.class, "vertex index");
	checkAssignable(classList, EDGE_PARAM_EDGE_INPUT_FORMAT_INDEX,
	edgeValueType(), EdgeInputFormat.class, "edge value");
	}
	}

	/**
	* If there is a vertex value combiner type, verify its
	* generic params match the job.
	*/
	private void verifyVertexValueCombinerGenericTypes() {
	Class<? extends VertexValueCombiner<V>> vertexValueCombiner =
	conf.getVertexValueCombinerClass();
	if (vertexValueCombiner != null) {
	Class<?>[] classList =
	getTypeArguments(VertexValueCombiner.class, vertexValueCombiner);
	checkAssignable(classList, VALUE_PARAM_VERTEX_VALUE_COMBINER_INDEX,
	vertexValueType(), VertexValueCombiner.class, "vertex value");
	}
	}

	/**
	* If there is a message combiner type, verify its
	* generic params match the job.
	*/
	private void verifyMessageCombinerGenericTypes() {
	MessageCombiner<I, M2> messageCombiner =
	conf.createOutgoingMessageCombiner();
	if (messageCombiner != null) {
	Class<?>[] classList =
	getTypeArguments(MessageCombiner.class, messageCombiner.getClass());
	checkEquals(classList, ID_PARAM_INDEX, vertexIndexType(),
	MessageCombiner.class, "vertex index");
	checkEquals(classList, MSG_COMBINER_PARAM_INDEX,
	outgoingMessageValueType(), MessageCombiner.class, "message value");
	}
	}

	/** Verify that the vertex output format's generic params match the job. */
	private void verifyVertexOutputFormatGenericTypes() {
	Class<? extends EdgeOutputFormat<I, V, E>>
	edgeOutputFormatClass = conf.getEdgeOutputFormatClass();
	if (conf.hasEdgeOutputFormat()) {
	Class<?>[] classList =
	getTypeArguments(EdgeOutputFormat.class, edgeOutputFormatClass);
	checkAssignable(classList, ID_PARAM_INDEX, vertexIndexType(),
	VertexOutputFormat.class, "vertex index");
	checkAssignable(classList, VALUE_PARAM_INDEX, vertexValueType(),
	VertexOutputFormat.class, "vertex value");
	checkAssignable(classList, EDGE_PARAM_INDEX, edgeValueType(),
	VertexOutputFormat.class, "edge value");
	}
	}

	/** Verify that the edge output format's generic params match the job. */
	private void verifyEdgeOutputFormatGenericTypes() {
	Class<? extends VertexOutputFormat<I, V, E>>
	vertexOutputFormatClass = conf.getVertexOutputFormatClass();
	if (conf.hasVertexOutputFormat()) {
	Class<?>[] classList =
	getTypeArguments(VertexOutputFormat.class, vertexOutputFormatClass);
	checkAssignable(classList, ID_PARAM_INDEX, vertexIndexType(),
	VertexOutputFormat.class, "vertex index");
	checkAssignable(classList, VALUE_PARAM_INDEX, vertexValueType(),
	VertexOutputFormat.class, "vertex value");
	checkAssignable(classList, EDGE_PARAM_INDEX, edgeValueType(),
	VertexOutputFormat.class, "edge value");
	}
	}

	/** Verify that the vertex value factory's type matches the job */
	private void verifyVertexValueFactoryGenericTypes() {
	Class<? extends VertexValueFactory<V>>
	vvfClass = conf.getVertexValueFactoryClass();
	if (DefaultVertexValueFactory.class.equals(vvfClass)) {
	return;
	}
	Class<?>[] classList = getTypeArguments(VertexValueFactory.class, vvfClass);
	checkEquals(classList, VALUE_PARAM_VERTEX_VALUE_FACTORY_INDEX,
	vertexValueType(), VertexValueFactory.class, "vertex value");
	}

	/**
	* If there is a vertex resolver,
	* validate the generic parameter types.
	* */
	private void verifyVertexResolverGenericTypes() {
	Class<? extends VertexResolver<I, V, E>>
	vrClass = conf.getVertexResolverClass();
	if (DefaultVertexResolver.class.equals(vrClass)) {
	return;
	}
	Class<?>[] classList =
	getTypeArguments(VertexResolver.class, vrClass);
	checkEquals(classList, ID_PARAM_INDEX, vertexIndexType(),
	VertexResolver.class, "vertex index");
	checkEquals(classList, VALUE_PARAM_INDEX, vertexValueType(),
	VertexResolver.class, "vertex value");
	checkEquals(classList, EDGE_PARAM_INDEX, edgeValueType(),
	VertexResolver.class, "edge value");
	}

	/**
	* Check that the type from computation equals the type from the class.
	*
	* @param classList classes from type
	* @param index array index of class to check
	* @param classFromComputation class from computation
	* @param klass Class type we're checking, only used for printing name
	* @param typeName Name of type we're checking
	*/
	private static void checkEquals(Class<?>[] classList, int index,
	Class<?> classFromComputation, Class klass, String typeName) {
	if (classList[index] == null) {
	LOG.warn(klass.getSimpleName() + " " + typeName + " type is not known");
	} else if (!classList[index].equals(classFromComputation)) {
	throw new IllegalStateException(
	"checkClassTypes: " + typeName + " types not equal, " +
	"computation - " + classFromComputation +
	", " + klass.getSimpleName() + " - " +
	classList[index]);
	}
	}

	/**
	* Check that the type from computation is assignable to type from the class.
	*
	* @param classList classes from type
	* @param index array index of class to check
	* @param classFromComputation class from computation
	* @param klass Class type we're checking, only used for printing name
	* @param typeName Name of type we're checking
	*/
	private static void checkAssignable(Class<?>[] classList, int index,
	Class<?> classFromComputation, Class klass, String typeName) {
	if (classList[index] == null) {
	LOG.warn(klass.getSimpleName() + " " + typeName + " type is not known");
	} else if (!classList[index].isAssignableFrom(classFromComputation)) {
	throw new IllegalStateException(
	"checkClassTypes: " + typeName + " types not assignable, " +
	"computation - " + classFromComputation +
	", " + klass.getSimpleName() + " - " +
	classList[EDGE_PARAM_EDGE_INPUT_FORMAT_INDEX]);
	}
	}
	}