private ExecutionVertex createVertex(final AbstractJobVertex jobVertex, final InstanceManager instanceManager,
			final ExecutionStage initialExecutionStage, final Configuration jobConfiguration)
			throws GraphConversionException {

		// If the user has requested instance type, check if the type is known by the current instance manager
		InstanceType instanceType = null;
		boolean userDefinedInstanceType = false;
		if (jobVertex.getInstanceType() != null) {

			userDefinedInstanceType = true;
			instanceType = instanceManager.getInstanceTypeByName(jobVertex.getInstanceType());
			if (instanceType == null) {
				throw new GraphConversionException("Requested instance type " + jobVertex.getInstanceType()
					+ " is not known to the instance manager");
			}
		}

		if (instanceType == null) {
			instanceType = instanceManager.getDefaultInstanceType();
		}

		// Create an initial execution vertex for the job vertex
		final Class<? extends AbstractInvokable> invokableClass = jobVertex.getInvokableClass();
		if (invokableClass == null) {
			throw new GraphConversionException("JobVertex " + jobVertex.getID() + " (" + jobVertex.getName()
				+ ") does not specify a task");
		}

		// Calculate the cryptographic signature of this vertex
		final ExecutionSignature signature = ExecutionSignature.createSignature(jobVertex.getInvokableClass(),
			jobVertex.getJobGraph().getJobID());

		// Create a group vertex for the job vertex

		ExecutionGroupVertex groupVertex = null;
		try {
			groupVertex = new ExecutionGroupVertex(jobVertex.getName(), jobVertex.getID(), this,
				jobVertex.getNumberOfSubtasks(), instanceType, userDefinedInstanceType,
				jobVertex.getNumberOfSubtasksPerInstance(), jobVertex.getVertexToShareInstancesWith() != null ? true
					: false, jobVertex.getNumberOfExecutionRetries(), jobVertex.getConfiguration(), signature,
				invokableClass);
		} catch (Throwable t) {
			throw new GraphConversionException(StringUtils.stringifyException(t));
		}

		// Run the configuration check the user has provided for the vertex
		try {
			jobVertex.checkConfiguration(groupVertex.getEnvironment().getInvokable());
		} catch (IllegalConfigurationException e) {
			throw new GraphConversionException(StringUtils.stringifyException(e));
		}

		// Check if the user's specifications for the number of subtasks are valid
		final int minimumNumberOfSubtasks = jobVertex.getMinimumNumberOfSubtasks(groupVertex.getEnvironment()
			.getInvokable());
		final int maximumNumberOfSubtasks = jobVertex.getMaximumNumberOfSubtasks(groupVertex.getEnvironment()
			.getInvokable());
		if (jobVertex.getNumberOfSubtasks() != -1) {
			if (jobVertex.getNumberOfSubtasks() < 1) {
				throw new GraphConversionException("Cannot split task " + jobVertex.getName() + " into "
					+ jobVertex.getNumberOfSubtasks() + " subtasks");
			}

			if (jobVertex.getNumberOfSubtasks() < minimumNumberOfSubtasks) {
				throw new GraphConversionException("Number of subtasks must be at least " + minimumNumberOfSubtasks);
			}

			if (maximumNumberOfSubtasks != -1) {
				if (jobVertex.getNumberOfSubtasks() > maximumNumberOfSubtasks) {
					throw new GraphConversionException("Number of subtasks for vertex " + jobVertex.getName()
						+ " can be at most " + maximumNumberOfSubtasks);
				}
			}
		}

		// Check number of subtasks per instance
		if (jobVertex.getNumberOfSubtasksPerInstance() != -1 && jobVertex.getNumberOfSubtasksPerInstance() < 1) {
			throw new GraphConversionException("Cannot set number of subtasks per instance to "
				+ jobVertex.getNumberOfSubtasksPerInstance() + " for vertex " + jobVertex.getName());
		}

		// Assign min/max to the group vertex (settings are actually applied in applyUserDefinedSettings)
		groupVertex.setMinMemberSize(minimumNumberOfSubtasks);
		groupVertex.setMaxMemberSize(maximumNumberOfSubtasks);

		// Register input and output vertices separately
		if (jobVertex instanceof AbstractJobInputVertex) {

			final InputSplit[] inputSplits;

			// let the task code compute the input splits
			if (groupVertex.getEnvironment().getInvokable() instanceof AbstractInputTask) {
				try {
					inputSplits = ((AbstractInputTask<?>) groupVertex.getEnvironment().getInvokable())
						.computeInputSplits(jobVertex.getNumberOfSubtasks());
				} catch (Exception e) {
					throw new GraphConversionException("Cannot compute input splits for " + groupVertex.getName(), e);
				}
			} else {
				throw new GraphConversionException("JobInputVertex contained a task class which was not an input task.");
			}

			if (inputSplits == null) {
				LOG.info("Job input vertex " + jobVertex.getName() + " generated 0 input splits");
			} else {
				LOG.info("Job input vertex " + jobVertex.getName() + " generated " + inputSplits.length
					+ " input splits");
			}

			// assign input splits
			groupVertex.setInputSplits(inputSplits);
		}
		// TODO: This is a quick workaround, problem can be solved in a more generic way
		if (jobVertex instanceof JobFileOutputVertex) {
			final JobFileOutputVertex jbov = (JobFileOutputVertex) jobVertex;
			jobVertex.getConfiguration().setString("outputPath", jbov.getFilePath().toString());
		}

		// Add group vertex to initial execution stage
		initialExecutionStage.addStageMember(groupVertex);

		final ExecutionVertex ev = new ExecutionVertex(this, groupVertex, jobVertex.getNumberOfForwardConnections(),
			jobVertex.getNumberOfBackwardConnections());

		// Assign initial instance to vertex (may be overwritten later on when user settings are applied)
		ev.setAllocatedResource(new AllocatedResource(DummyInstance.createDummyInstance(instanceType), instanceType,
			null));

		return ev;
	}
private ExecutionVertex createVertex ( final AbstractJobVertex jobVertex , final InstanceManager instanceManager , final ExecutionStage initialExecutionStage , final Configuration jobConfiguration ) throws GraphConversionException { InstanceType instanceType = null ; boolean userDefinedInstanceType = false ; if ( jobVertex . getInstanceType ( ) != null ) { userDefinedInstanceType = true ; instanceType = instanceManager . getInstanceTypeByName ( jobVertex . getInstanceType ( ) ) ; if ( instanceType == null ) { throw new GraphConversionException ( "Requested instance type " + jobVertex . getInstanceType ( ) + " is not known to the instance manager" ) ; } } if ( instanceType == null ) { instanceType = instanceManager . getDefaultInstanceType ( ) ; } final Class < ? extends AbstractInvokable > invokableClass = jobVertex . getInvokableClass ( ) ; if ( invokableClass == null ) { throw new GraphConversionException ( "JobVertex " + jobVertex . getID ( ) + " (" + jobVertex . getName ( ) + ") does not specify a task" ) ; } final ExecutionSignature signature = ExecutionSignature . createSignature ( jobVertex . getInvokableClass ( ) , jobVertex . getJobGraph ( ) . getJobID ( ) ) ; ExecutionGroupVertex groupVertex = null ; try { groupVertex = new ExecutionGroupVertex ( jobVertex . getName ( ) , jobVertex . getID ( ) , this , jobVertex . getNumberOfSubtasks ( ) , instanceType , userDefinedInstanceType , jobVertex . getNumberOfSubtasksPerInstance ( ) , jobVertex . getVertexToShareInstancesWith ( ) != null ? true : false , jobVertex . getNumberOfExecutionRetries ( ) , jobVertex . getConfiguration ( ) , signature , invokableClass ) ; } catch ( Throwable t ) { throw new GraphConversionException ( StringUtils . stringifyException ( t ) ) ; } try { jobVertex . checkConfiguration ( groupVertex . getEnvironment ( ) . getInvokable ( ) ) ; } catch ( IllegalConfigurationException e ) { throw new GraphConversionException ( StringUtils . stringifyException ( e ) ) ; } final int minimumNumberOfSubtasks = jobVertex . getMinimumNumberOfSubtasks ( groupVertex . getEnvironment ( ) . getInvokable ( ) ) ; final int maximumNumberOfSubtasks = jobVertex . getMaximumNumberOfSubtasks ( groupVertex . getEnvironment ( ) . getInvokable ( ) ) ; if ( jobVertex . getNumberOfSubtasks ( ) != - 1 ) { if ( jobVertex . getNumberOfSubtasks ( ) < 1 ) { throw new GraphConversionException ( "Cannot split task " + jobVertex . getName ( ) + " into " + jobVertex . getNumberOfSubtasks ( ) + " subtasks" ) ; } if ( jobVertex . getNumberOfSubtasks ( ) < minimumNumberOfSubtasks ) { throw new GraphConversionException ( "Number of subtasks must be at least " + minimumNumberOfSubtasks ) ; } if ( maximumNumberOfSubtasks != - 1 ) { if ( jobVertex . getNumberOfSubtasks ( ) > maximumNumberOfSubtasks ) { throw new GraphConversionException ( "Number of subtasks for vertex " + jobVertex . getName ( ) + " can be at most " + maximumNumberOfSubtasks ) ; } } } if ( jobVertex . getNumberOfSubtasksPerInstance ( ) != - 1 && jobVertex . getNumberOfSubtasksPerInstance ( ) < 1 ) { throw new GraphConversionException ( "Cannot set number of subtasks per instance to " + jobVertex . getNumberOfSubtasksPerInstance ( ) + " for vertex " + jobVertex . getName ( ) ) ; } groupVertex . setMinMemberSize ( minimumNumberOfSubtasks ) ; groupVertex . setMaxMemberSize ( maximumNumberOfSubtasks ) ; if ( jobVertex instanceof AbstractJobInputVertex ) { final InputSplit [ ] inputSplits ; if ( groupVertex . getEnvironment ( ) . getInvokable ( ) instanceof AbstractInputTask ) { try { inputSplits = ( ( AbstractInputTask < ? > ) groupVertex . getEnvironment ( ) . getInvokable ( ) ) . computeInputSplits ( jobVertex . getNumberOfSubtasks ( ) ) ; } catch ( Exception e ) { throw new GraphConversionException ( "Cannot compute input splits for " + groupVertex . getName ( ) , e ) ; } } else { throw new GraphConversionException ( "JobInputVertex contained a task class which was not an input task." ) ; } if ( inputSplits == null ) { LOG . info ( "Job input vertex " + jobVertex . getName ( ) + " generated 0 input splits" ) ; } else { LOG . info ( "Job input vertex " + jobVertex . getName ( ) + " generated " + inputSplits . length + " input splits" ) ; } groupVertex . setInputSplits ( inputSplits ) ; } if ( jobVertex instanceof JobFileOutputVertex ) { final JobFileOutputVertex jbov = ( JobFileOutputVertex ) jobVertex ; jobVertex . getConfiguration ( ) . setString ( "outputPath" , jbov . getFilePath ( ) . toString ( ) ) ; } initialExecutionStage . addStageMember ( groupVertex ) ; final ExecutionVertex ev = new ExecutionVertex ( this , groupVertex , jobVertex . getNumberOfForwardConnections ( ) , jobVertex . getNumberOfBackwardConnections ( ) ) ; ev . setAllocatedResource ( new AllocatedResource ( DummyInstance . createDummyInstance ( instanceType ) , instanceType , null ) ) ; return ev ; }
Creates an execution vertex from a job vertex.

@param jobVertex
the job vertex to create the execution vertex from
@param instanceManager
the instanceManager
@param initialExecutionStage
the initial execution stage all group vertices are added to
@param jobConfiguration
the configuration object originally attached to the {@link JobGraph}
@return the new execution vertex
@throws GraphConversionException
thrown if the job vertex is of an unknown subclass