code: public static ExecutionState getAggregateJobVertexState(int[] verticesPerState, int parallelism) {
  		if (verticesPerState == null || verticesPerState.length != ExecutionState.values().length) {
  			throw new IllegalArgumentException("Must provide an array as large as there are execution states.");
  		}
  
  		if (verticesPerState[ExecutionState.FAILED.ordinal()] > 0) {
  			return ExecutionState.FAILED;
  		}
  		if (verticesPerState[ExecutionState.CANCELING.ordinal()] > 0) {
  			return ExecutionState.CANCELING;
  		}
  		else if (verticesPerState[ExecutionState.CANCELED.ordinal()] > 0) {
  			return ExecutionState.CANCELED;
  		}
  		else if (verticesPerState[ExecutionState.RUNNING.ordinal()] > 0) {
  			return ExecutionState.RUNNING;
  		}
  		else if (verticesPerState[ExecutionState.FINISHED.ordinal()] > 0) {
  			return verticesPerState[ExecutionState.FINISHED.ordinal()] == parallelism ?
  					ExecutionState.FINISHED : ExecutionState.RUNNING;
  		}
  		else {
  			// all else collapses under created
  			return ExecutionState.CREATED;
  		}
  	}
docs: A utility function that computes an "aggregated" state for the vertex.
  
  <p>This state is not used anywhere in the  coordination, but can be used for display
  in dashboards to as a summary for how the particular parallel operation represented by
  this ExecutionJobVertex is currently behaving.
  
  <p>For example, if at least one parallel task is failed, the aggregate state is failed.
  If not, and at least one parallel task is cancelling (or cancelled), the aggregate state
  is cancelling (or cancelled). If all tasks are finished, the aggregate state is finished,
  and so on.
  
  @param verticesPerState The number of vertices in each state (indexed by the ordinal of
  the ExecutionState values).
  @param parallelism The parallelism of the ExecutionJobVertex
  
  @return The aggregate state of this ExecutionJobVertex.