docs: public Execution resetForNewExecution(final long timestamp, final long originatingGlobalModVersion)
  			throws GlobalModVersionMismatch {
  		LOG.debug("Resetting execution vertex {} for new execution.", getTaskNameWithSubtaskIndex());
  
  		synchronized (priorExecutions) {
  			// check if another global modification has been triggered since the
  			// action that originally caused this reset/restart happened
  			final long actualModVersion = getExecutionGraph().getGlobalModVersion();
  			if (actualModVersion > originatingGlobalModVersion) {
  				// global change happened since, reject this action
  				throw new GlobalModVersionMismatch(originatingGlobalModVersion, actualModVersion);
  			}
  
  			final Execution oldExecution = currentExecution;
  			final ExecutionState oldState = oldExecution.getState();
  
  			if (oldState.isTerminal()) {
  				priorExecutions.add(oldExecution.archive());
  
  				final Execution newExecution = new Execution(
  					getExecutionGraph().getFutureExecutor(),
  					this,
  					oldExecution.getAttemptNumber() + 1,
  					originatingGlobalModVersion,
  					timestamp,
  					timeout);
  
  				currentExecution = newExecution;
  
  				synchronized (inputSplits) {
  					InputSplitAssigner assigner = jobVertex.getSplitAssigner();
  					if (assigner != null) {
  						assigner.returnInputSplit(inputSplits, getParallelSubtaskIndex());
  						inputSplits.clear();
  					}
  				}
  
  				CoLocationGroup grp = jobVertex.getCoLocationGroup();
  				if (grp != null) {
  					locationConstraint = grp.getLocationConstraint(subTaskIndex);
  				}
  
  				// register this execution at the execution graph, to receive call backs
  				getExecutionGraph().registerExecution(newExecution);
  
  				// if the execution was 'FINISHED' before, tell the ExecutionGraph that
  				// we take one step back on the road to reaching global FINISHED
  				if (oldState == FINISHED) {
  					getExecutionGraph().vertexUnFinished();
  				}
  
  				// reset the intermediate results
  				for (IntermediateResultPartition resultPartition : resultPartitions.values()) {
  					resultPartition.resetForNewExecution();
  				}
  
  				return newExecution;
  			}
  			else {
  				throw new IllegalStateException("Cannot reset a vertex that is in non-terminal state " + oldState);
  			}
  		}
  	}