code: @Override
  	public BufferProvider getBufferProvider(JobID jobID, ChannelID sourceChannelID) throws IOException {
  		EnvelopeReceiverList receiverList = getReceiverList(jobID, sourceChannelID, false);
  		
  		// check if the receiver is already gone
  		if (receiverList == null) {
  			return this.discardBufferPool;
  		}
  
  		if (!receiverList.hasLocalReceiver() || receiverList.hasRemoteReceiver()) {
  			throw new IOException("The destination to be looked up is not a single local endpoint.");
  		}
  		
  
  		ChannelID localReceiver = receiverList.getLocalReceiver();
  		Channel channel = this.channels.get(localReceiver);
  		
  		if (channel == null) {
  			// receiver is already canceled
  			return this.discardBufferPool;
  		}
  
  		if (!channel.isInputChannel()) {
  			throw new IOException("Channel context for local receiver " + localReceiver + " is not an input channel context");
  		}
  
  		return (InputChannel<?>) channel;
  	}
docs: -----------------------------------------------------------------------------------------------------------------