code: @SuppressWarnings("unchecked")
  	private void initInputReaders() throws Exception {
  		int numGates = 0;
  		//  ---------------- create the input readers ---------------------
  		// in case where a logical input unions multiple physical inputs, create a union reader
  		final int groupSize = this.config.getGroupSize(0);
  		numGates += groupSize;
  		if (groupSize == 1) {
  			// non-union case
  			inputReader = new MutableRecordReader<DeserializationDelegate<IT>>(
  					getEnvironment().getInputGate(0),
  					getEnvironment().getTaskManagerInfo().getTmpDirectories());
  		} else if (groupSize > 1){
  			// union case
  			inputReader = new MutableRecordReader<IOReadableWritable>(
  					new UnionInputGate(getEnvironment().getAllInputGates()),
  					getEnvironment().getTaskManagerInfo().getTmpDirectories());
  		} else {
  			throw new Exception("Illegal input group size in task configuration: " + groupSize);
  		}
  		
  		this.inputTypeSerializerFactory = this.config.getInputSerializer(0, getUserCodeClassLoader());
  		@SuppressWarnings({ "rawtypes" })
  		final MutableObjectIterator<?> iter = new ReaderIterator(inputReader, this.inputTypeSerializerFactory.getSerializer());
  		this.reader = (MutableObjectIterator<IT>)iter;
  
  		// final sanity check
  		if (numGates != this.config.getNumInputs()) {
  			throw new Exception("Illegal configuration: Number of input gates and group sizes are not consistent.");
  		}
  	}
docs: Initializes the input readers of the DataSinkTask.
  
  @throws RuntimeException
  Thrown in case of invalid task input configuration.