code: public InputSplit getNextInputSplit(final ExecutionVertex vertex, final int sequenceNumber) {
  
  		InputSplit nextInputSplit = this.inputSplitTracker.getInputSplitFromLog(vertex, sequenceNumber);
  		if (nextInputSplit != null) {
  			LOG.info("Input split " + nextInputSplit.getSplitNumber() + " for vertex " + vertex + " replayed from log");
  			return nextInputSplit;
  		}
  
  		final ExecutionGroupVertex groupVertex = vertex.getGroupVertex();
  		final InputSplitAssigner inputSplitAssigner = this.assignerCache.get(groupVertex);
  		if (inputSplitAssigner == null) {
  			final JobID jobID = groupVertex.getExecutionStage().getExecutionGraph().getJobID();
  			LOG.error("Cannot find input assigner for group vertex " + groupVertex.getName() + " (job " + jobID + ")");
  			return null;
  		}
  
  		nextInputSplit = inputSplitAssigner.getNextInputSplit(vertex);
  		if (nextInputSplit != null) {
  			this.inputSplitTracker.addInputSplitToLog(vertex, sequenceNumber, nextInputSplit);
  			LOG.info(vertex + " receives input split " + nextInputSplit.getSplitNumber());
  		}
  
  		return nextInputSplit;
  	}
docs: Returns the next input split the input split manager (or the responsible {@link InputSplitAssigner} to be more
  precise) has chosen for the given vertex to consume.
  
  @param vertex
  the vertex for which the next input split is to be determined
  @param sequenceNumber
  the sequence number of the vertex's request
  @return the next input split to consume or <code>null</code> if the vertex shall consume no more input splits