docs: 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);
  			}
  		}
  	}