docs: protected void initBroadcastInputReaders() throws Exception {
  		final int numBroadcastInputs = this.config.getNumBroadcastInputs();
  		final MutableReader<?>[] broadcastInputReaders = new MutableReader<?>[numBroadcastInputs];
  
  		int currentReaderOffset = config.getNumInputs();
  
  		for (int i = 0; i < this.config.getNumBroadcastInputs(); i++) {
  			//  ---------------- create the input readers ---------------------
  			// in case where a logical input unions multiple physical inputs, create a union reader
  			final int groupSize = this.config.getBroadcastGroupSize(i);
  			if (groupSize == 1) {
  				// non-union case
  				broadcastInputReaders[i] = new MutableRecordReader<IOReadableWritable>(
  						getEnvironment().getInputGate(currentReaderOffset),
  						getEnvironment().getTaskManagerInfo().getTmpDirectories());
  			} else if (groupSize > 1){
  				// union case
  				InputGate[] readers = new InputGate[groupSize];
  				for (int j = 0; j < groupSize; ++j) {
  					readers[j] = getEnvironment().getInputGate(currentReaderOffset + j);
  				}
  				broadcastInputReaders[i] = new MutableRecordReader<IOReadableWritable>(
  						new UnionInputGate(readers),
  						getEnvironment().getTaskManagerInfo().getTmpDirectories());
  			} else {
  				throw new Exception("Illegal input group size in task configuration: " + groupSize);
  			}
  
  			currentReaderOffset += groupSize;
  		}
  		this.broadcastInputReaders = broadcastInputReaders;
  	}