private void reset ( long globalModVersionOfFailover ) { try { // reset all connected ExecutionVertexes final Collection < CoLocationGroup > colGroups = new HashSet <> ( ) ; final long restartTimestamp = System . currentTimeMillis ( ) ; for ( ExecutionVertex ev : connectedExecutionVertexes ) { CoLocationGroup cgroup = ev . getJobVertex ( ) . getCoLocationGroup ( ) ; if ( cgroup != null && ! colGroups . contains ( cgroup ) ) { cgroup . resetConstraints ( ) ; colGroups . add ( cgroup ) ; } ev . resetForNewExecution ( restartTimestamp , globalModVersionOfFailover ) ; } if ( transitionState ( JobStatus . CANCELED , JobStatus . CREATED ) ) { restart ( globalModVersionOfFailover ) ; } else { LOG . info ( "FailoverRegion {} switched from CANCELLING to CREATED fail, will fail this region again." , id ) ; failover ( globalModVersionOfFailover ) ; } } catch ( GlobalModVersionMismatch e ) { // happens when a global recovery happens concurrently to the regional recovery // go back to a clean state state = JobStatus . RUNNING ; } catch ( Throwable e ) { LOG . info ( "FailoverRegion {} reset fail, will failover again." , id ) ; failover ( globalModVersionOfFailover ) ; } }
reset all executions in this sub graph