code: @Override
  	public void addBufferConsumer(BufferConsumer bufferConsumer, int subpartitionIndex) throws IOException {
  		checkNotNull(bufferConsumer);
  
  		ResultSubpartition subpartition;
  		try {
  			checkInProduceState();
  			subpartition = subpartitions[subpartitionIndex];
  		}
  		catch (Exception ex) {
  			bufferConsumer.close();
  			throw ex;
  		}
  
  		if (subpartition.add(bufferConsumer)) {
  			notifyPipelinedConsumers();
  		}
  	}
docs: ------------------------------------------------------------------------