public IndexRecord getIndexInformation(String mapId, int reduce,
                                         Path fileName, String expectedIndexOwner)
    throws IOException {

    IndexInformation info = cache.get(mapId);

    if (info == null) {
      info = readIndexFileToCache(fileName, mapId, expectedIndexOwner);
    } else {
      synchronized(info) {
        while (isUnderConstruction(info)) {
          try {
            info.wait();
          } catch (InterruptedException e) {
            throw new IOException("Interrupted waiting for construction", e);
          }
        }
      }
      LOG.debug("IndexCache HIT: MapId " + mapId + " found");
    }

    if (info.mapSpillRecord.size() == 0 ||
        info.mapSpillRecord.size() <= reduce) {
      throw new IOException("Invalid request " +
        " Map Id = " + mapId + " Reducer = " + reduce +
        " Index Info Length = " + info.mapSpillRecord.size());
    }
    return info.mapSpillRecord.getIndex(reduce);
  }
@Test
  public void testInvalidReduceNumberOrLength() throws Exception {
    fs.delete(p, true);
    conf.setInt(MRJobConfig.SHUFFLE_INDEX_CACHE, 1);
    final int partsPerMap = 1000;
    final int bytesPerFile = partsPerMap * 24;
    IndexCache cache = new IndexCache(conf);

    // fill cache
    Path feq = new Path(p, "invalidReduceOrPartsPerMap");
    writeFile(fs, feq, bytesPerFile, partsPerMap);

    // Number of reducers should always be less than partsPerMap as reducer
    // numbers start from 0 and there cannot be more reducer than parts

    try {
      // Number of reducers equal to partsPerMap
      cache.getIndexInformation("reduceEqualPartsPerMap", 
               partsPerMap, // reduce number == partsPerMap
               feq, UserGroupInformation.getCurrentUser().getShortUserName());
      fail("Number of reducers equal to partsPerMap did not fail");
    } catch (Exception e) {
      if (!(e instanceof IOException)) {
        throw e;
      }
    }

    try {
      // Number of reducers more than partsPerMap
      cache.getIndexInformation(
      "reduceMorePartsPerMap", 
      partsPerMap + 1, // reduce number > partsPerMap
      feq, UserGroupInformation.getCurrentUser().getShortUserName());
      fail("Number of reducers more than partsPerMap did not fail");
    } catch (Exception e) {
      if (!(e instanceof IOException)) {
        throw e;
      }
    }
  }