code: @Override
  	public BufferOrEvent getNextNonBlocked() throws Exception {
  		while (true) {
  			// process buffered BufferOrEvents before grabbing new ones
  			Optional<BufferOrEvent> next;
  			if (currentBuffered == null) {
  				next = inputGate.getNextBufferOrEvent();
  			}
  			else {
  				next = Optional.ofNullable(currentBuffered.getNext());
  				if (!next.isPresent()) {
  					completeBufferedSequence();
  					return getNextNonBlocked();
  				}
  			}
  
  			if (!next.isPresent()) {
  				if (!endOfStream) {
  					// end of input stream. stream continues with the buffered data
  					endOfStream = true;
  					releaseBlocksAndResetBarriers();
  					return getNextNonBlocked();
  				}
  				else {
  					// final end of both input and buffered data
  					return null;
  				}
  			}
  
  			BufferOrEvent bufferOrEvent = next.get();
  			if (isBlocked(bufferOrEvent.getChannelIndex())) {
  				// if the channel is blocked, we just store the BufferOrEvent
  				bufferBlocker.add(bufferOrEvent);
  				checkSizeLimit();
  			}
  			else if (bufferOrEvent.isBuffer()) {
  				return bufferOrEvent;
  			}
  			else if (bufferOrEvent.getEvent().getClass() == CheckpointBarrier.class) {
  				if (!endOfStream) {
  					// process barriers only if there is a chance of the checkpoint completing
  					processBarrier((CheckpointBarrier) bufferOrEvent.getEvent(), bufferOrEvent.getChannelIndex());
  				}
  			}
  			else if (bufferOrEvent.getEvent().getClass() == CancelCheckpointMarker.class) {
  				processCancellationBarrier((CancelCheckpointMarker) bufferOrEvent.getEvent());
  			}
  			else {
  				if (bufferOrEvent.getEvent().getClass() == EndOfPartitionEvent.class) {
  					processEndOfPartition();
  				}
  				return bufferOrEvent;
  			}
  		}
  	}
docs: ------------------------------------------------------------------------