private void generateAllFailoverRegion ( List < ExecutionJobVertex > newJobVerticesTopological ) { final IdentityHashMap < ExecutionVertex , ArrayList < ExecutionVertex > > vertexToRegion = new IdentityHashMap <> ( ) ; // we use the map (list -> null) to imitate an IdentityHashSet (which does not exist) final IdentityHashMap < ArrayList < ExecutionVertex > , Object > distinctRegions = new IdentityHashMap <> ( ) ; // this loop will worst case iterate over every edge in the graph (complexity is O(#edges)) for ( ExecutionJobVertex ejv : newJobVerticesTopological ) { // currently, jobs with a co-location constraint fail as one // we want to improve that in the future (or get rid of co-location constraints) if ( ejv . getCoLocationGroup ( ) != null ) { makeAllOneRegion ( newJobVerticesTopological ) ; return ; } // see if this JobVertex one has pipelined inputs at all final List < IntermediateResult > inputs = ejv . getInputs ( ) ; final int numInputs = inputs . size ( ) ; boolean hasPipelinedInputs = false ; for ( IntermediateResult input : inputs ) { if ( input . getResultType ( ) . isPipelined ( ) ) { hasPipelinedInputs = true ; break ; } } if ( hasPipelinedInputs ) { // build upon the predecessors for ( ExecutionVertex ev : ejv . getTaskVertices ( ) ) { // remember the region in which we are ArrayList < ExecutionVertex > thisRegion = null ; for ( int inputNum = 0 ; inputNum < numInputs ; inputNum ++ ) { if ( inputs . get ( inputNum ) . getResultType ( ) . isPipelined ( ) ) { for ( ExecutionEdge edge : ev . getInputEdges ( inputNum ) ) { final ExecutionVertex predecessor = edge . getSource ( ) . getProducer ( ) ; final ArrayList < ExecutionVertex > predecessorRegion = vertexToRegion . get ( predecessor ) ; if ( thisRegion != null ) { // we already have a region. see if it is the same as the predecessor's region if ( predecessorRegion != thisRegion ) { // we need to merge our region and the predecessor's region predecessorRegion . addAll ( thisRegion ) ; distinctRegions . remove ( thisRegion ) ; thisRegion = predecessorRegion ; // remap the vertices from that merged region for ( ExecutionVertex inPredRegion : predecessorRegion ) { vertexToRegion . put ( inPredRegion , thisRegion ) ; } } } else if ( predecessor != null ) { // first case, make this our region thisRegion = predecessorRegion ; thisRegion . add ( ev ) ; vertexToRegion . put ( ev , thisRegion ) ; } else { // throw an uncaught exception here // this is a bug and not a recoverable situation throw new FlinkRuntimeException ( "bug in the logic to construct the pipelined failover regions" ) ; } } } } } } else { // no pipelined inputs, start a new region for ( ExecutionVertex ev : ejv . getTaskVertices ( ) ) { ArrayList < ExecutionVertex > region = new ArrayList <> ( 1 ) ; region . add ( ev ) ; vertexToRegion . put ( ev , region ) ; distinctRegions . put ( region , null ) ; } } } // now that we have all regions, create the failover region objects LOG . info ( "Creating {} individual failover regions for job {} ({})" , distinctRegions . size ( ) , executionGraph . getJobName ( ) , executionGraph . getJobID ( ) ) ; for ( List < ExecutionVertex > region : distinctRegions . keySet ( ) ) { final FailoverRegion failoverRegion = createFailoverRegion ( executionGraph , region ) ; for ( ExecutionVertex ev : region ) { this . vertexToRegion . put ( ev , failoverRegion ) ; } } }
Generate all the FailoverRegion from the new added job vertexes