code: public ChannelFuture requestSubpartition(
  			final ResultPartitionID partitionId,
  			final int subpartitionIndex,
  			final RemoteInputChannel inputChannel,
  			int delayMs) throws IOException {
  
  		checkNotClosed();
  
  		LOG.debug("Requesting subpartition {} of partition {} with {} ms delay.",
  				subpartitionIndex, partitionId, delayMs);
  
  		clientHandler.addInputChannel(inputChannel);
  
  		final PartitionRequest request = new PartitionRequest(
  				partitionId, subpartitionIndex, inputChannel.getInputChannelId(), inputChannel.getInitialCredit());
  
  		final ChannelFutureListener listener = new ChannelFutureListener() {
  			@Override
  			public void operationComplete(ChannelFuture future) throws Exception {
  				if (!future.isSuccess()) {
  					clientHandler.removeInputChannel(inputChannel);
  					SocketAddress remoteAddr = future.channel().remoteAddress();
  					inputChannel.onError(
  							new LocalTransportException(
  								String.format("Sending the partition request to '%s' failed.", remoteAddr),
  								future.channel().localAddress(), future.cause()
  							));
  				}
  			}
  		};
  
  		if (delayMs == 0) {
  			ChannelFuture f = tcpChannel.writeAndFlush(request);
  			f.addListener(listener);
  			return f;
  		} else {
  			final ChannelFuture[] f = new ChannelFuture[1];
  			tcpChannel.eventLoop().schedule(new Runnable() {
  				@Override
  				public void run() {
  					f[0] = tcpChannel.writeAndFlush(request);
  					f[0].addListener(listener);
  				}
  			}, delayMs, TimeUnit.MILLISECONDS);
  
  			return f[0];
  		}
  	}
docs: Requests a remote intermediate result partition queue.
  
  <p>The request goes to the remote producer, for which this partition
  request client instance has been created.