private void applyUserDefinedSettings ( final HashMap < AbstractJobVertex , ExecutionGroupVertex > temporaryGroupVertexMap ) throws GraphConversionException { // The check for cycles in the dependency chain for instance sharing is already checked in // <code>submitJob</code> method of the job manager // If there is no cycle, apply the settings to the corresponding group vertices final Iterator < Map . Entry < AbstractJobVertex , ExecutionGroupVertex > > it = temporaryGroupVertexMap . entrySet ( ) . iterator ( ) ; while ( it . hasNext ( ) ) { final Map . Entry < AbstractJobVertex , ExecutionGroupVertex > entry = it . next ( ) ; final AbstractJobVertex jobVertex = entry . getKey ( ) ; if ( jobVertex . getVertexToShareInstancesWith ( ) != null ) { final AbstractJobVertex vertexToShareInstancesWith = jobVertex . getVertexToShareInstancesWith ( ) ; final ExecutionGroupVertex groupVertex = entry . getValue ( ) ; final ExecutionGroupVertex groupVertexToShareInstancesWith = temporaryGroupVertexMap . get ( vertexToShareInstancesWith ) ; groupVertex . shareInstancesWith ( groupVertexToShareInstancesWith ) ; } } // Second, we create the number of execution vertices each group vertex is supposed to manage Iterator < ExecutionGroupVertex > it2 = new ExecutionGroupVertexIterator ( this , true , - 1 ) ; while ( it2 . hasNext ( ) ) { final ExecutionGroupVertex groupVertex = it2 . next ( ) ; if ( groupVertex . isNumberOfMembersUserDefined ( ) ) { groupVertex . createInitialExecutionVertices ( groupVertex . getUserDefinedNumberOfMembers ( ) ) ; groupVertex . repairSubtasksPerInstance ( ) ; } } // Finally, apply the channel settings channel settings it2 = new ExecutionGroupVertexIterator ( this , true , - 1 ) ; while ( it2 . hasNext ( ) ) { final ExecutionGroupVertex groupVertex = it2 . next ( ) ; for ( int i = 0 ; i < groupVertex . getNumberOfForwardLinks ( ) ; i ++ ) { final ExecutionGroupEdge edge = groupVertex . getForwardEdge ( i ) ; if ( edge . isChannelTypeUserDefined ( ) ) { edge . changeChannelType ( edge . getChannelType ( ) ) ; } // Create edges between execution vertices createExecutionEdgesForGroupEdge ( edge ) ; } } // Repair the instance assignment after having changed the channel types repairInstanceAssignment ( ) ; // Repair the instance sharing among different group vertices repairInstanceSharing ( ) ; // Finally, repair the stages repairStages ( ) ; }
Applies the user defined settings to the execution graph .