code: @SuppressWarnings("unchecked")
  	private void initInputReaders() throws Exception {
  		
  		MutableReader<?> inputReader;
  		
  		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>>(this);
  		} else if (groupSize > 1){
  			// union case
  			
  			MutableRecordReader<IOReadableWritable>[] readers = new MutableRecordReader[groupSize];
  			for (int j = 0; j < groupSize; ++j) {
  				readers[j] = new MutableRecordReader<IOReadableWritable>(this);
  			}
  			inputReader = new MutableUnionRecordReader<IOReadableWritable>(readers);
  		} else {
  			throw new Exception("Illegal input group size in task configuration: " + groupSize);
  		}
  		
  		this.inputTypeSerializerFactory = this.config.getInputSerializer(0, this.userCodeClassLoader);
  		
  		if (this.inputTypeSerializerFactory.getDataType() == Record.class) {
  			// record specific deserialization
  			MutableReader<Record> reader = (MutableReader<Record>) inputReader;
  			this.reader = (MutableObjectIterator<IT>)new RecordReaderIterator(reader);
  		} else {
  			// generic data type serialization
  			MutableReader<DeserializationDelegate<?>> reader = (MutableReader<DeserializationDelegate<?>>) inputReader;
  			@SuppressWarnings({ "rawtypes" })
  			final MutableObjectIterator<?> iter = new ReaderIterator(reader, 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.