code: @Override
  	protected MemorySegment nextSegment(MemorySegment current) throws IOException {
  		// check if we are at our end
  		if (this.inLastBlock) {
  			throw new EOFException();
  		}
  
  		// send a request first. if we have only a single segment, this same segment will be the one obtained in
  		// the next lines
  		if (current != null) {
  			sendReadRequest(current);
  		}
  
  		// get the next segment
  		final MemorySegment seg = this.reader.getNextReturnedBlock();
  
  		// check the header
  		if (seg.getShort(0) != ChannelWriterOutputView.HEADER_MAGIC_NUMBER) {
  			throw new IOException("The current block does not belong to a ChannelWriterOutputView / " +
  					"ChannelReaderInputView: Wrong magic number.");
  		}
  		if ( (seg.getShort(ChannelWriterOutputView.HEADER_FLAGS_OFFSET) & ChannelWriterOutputView.FLAG_LAST_BLOCK) != 0) {
  			// last block
  			this.numRequestsRemaining = 0;
  			this.inLastBlock = true;
  		}
  
  		return seg;
  	}
docs: Gets the next segment from the asynchronous block reader. If more requests are to be issued, the method
  first sends a new request with the current memory segment. If no more requests are pending, the method
  adds the segment to the readers return queue, which thereby effectively collects all memory segments.
  Secondly, the method fetches the next non-consumed segment
  returned by the reader. If no further segments are available, this method thrown an {@link EOFException}.
  
  @param current The memory segment used for the next request.
  @return The memory segment to read from next.
  
  @throws EOFException Thrown, if no further segments are available.
  @throws IOException Thrown, if an I/O error occurred while reading
  @see AbstractPagedInputView#nextSegment(org.apache.flink.core.memory.MemorySegment)