private void createInitialGroupEdges ( final HashMap < AbstractJobVertex , ExecutionVertex > vertexMap ) throws GraphConversionException { Iterator < Map . Entry < AbstractJobVertex , ExecutionVertex > > it = vertexMap . entrySet ( ) . iterator ( ) ; while ( it . hasNext ( ) ) { final Map . Entry < AbstractJobVertex , ExecutionVertex > entry = it . next ( ) ; final AbstractJobVertex sjv = entry . getKey ( ) ; final ExecutionVertex sev = entry . getValue ( ) ; final ExecutionGroupVertex sgv = sev . getGroupVertex ( ) ; // First compare number of output gates if ( sjv . getNumberOfForwardConnections ( ) != sgv . getEnvironment ( ) . getNumberOfOutputGates ( ) ) { throw new GraphConversionException ( "Job and execution vertex " + sjv . getName ( ) + " have different number of outputs" ) ; } if ( sjv . getNumberOfBackwardConnections ( ) != sgv . getEnvironment ( ) . getNumberOfInputGates ( ) ) { throw new GraphConversionException ( "Job and execution vertex " + sjv . getName ( ) + " have different number of inputs" ) ; } // First, build the group edges for ( int i = 0 ; i < sjv . getNumberOfForwardConnections ( ) ; ++ i ) { final JobEdge edge = sjv . getForwardConnection ( i ) ; final AbstractJobVertex tjv = edge . getConnectedVertex ( ) ; final ExecutionVertex tev = vertexMap . get ( tjv ) ; final ExecutionGroupVertex tgv = tev . getGroupVertex ( ) ; // Use NETWORK as default channel type if nothing else is defined by the user ChannelType channelType = edge . getChannelType ( ) ; boolean userDefinedChannelType = true ; if ( channelType == null ) { userDefinedChannelType = false ; channelType = ChannelType . NETWORK ; } final DistributionPattern distributionPattern = edge . getDistributionPattern ( ) ; // Connect the corresponding group vertices and copy the user settings from the job edge final ExecutionGroupEdge groupEdge = sgv . wireTo ( tgv , edge . getIndexOfInputGate ( ) , i , channelType , userDefinedChannelType , distributionPattern ) ; final ExecutionGate outputGate = new ExecutionGate ( new GateID ( ) , sev , groupEdge , false ) ; sev . insertOutputGate ( i , outputGate ) ; final ExecutionGate inputGate = new ExecutionGate ( new GateID ( ) , tev , groupEdge , true ) ; tev . insertInputGate ( edge . getIndexOfInputGate ( ) , inputGate ) ; } } }
Creates the initial edges between the group vertices