IndexCache { public void removeMap(String mapId) { IndexInformation info = cache.get(mapId); if (info == null || isUnderConstruction(info)) { return; } info = cache.remove(mapId); if (info != null) { totalMemoryUsed.addAndGet(-info.getSize()); if (!queue.remove(mapId)) { LOG.warn("Map ID" + mapId + " not found in queue!!"); } } else { LOG.info("Map ID " + mapId + " not found in cache"); } } IndexCache(JobConf conf); IndexRecord getIndexInformation(String mapId, int reduce,
                                         Path fileName, String expectedIndexOwner); void removeMap(String mapId);  }
@Test public void testRemoveMap() throws Exception { fs.delete(p, true); conf.setInt(TTConfig.TT_INDEX_CACHE, 10); final int partsPerMap = 100000; final int bytesPerFile = partsPerMap * 24; final IndexCache cache = new IndexCache(conf); final Path big = new Path(p, "bigIndex"); final String user = UserGroupInformation.getCurrentUser().getShortUserName(); writeFile(fs, big, bytesPerFile, partsPerMap); for (int i = 0; i < 20; ++i) { Thread getInfoThread = new Thread() { @Override public void run() { try { cache.getIndexInformation("bigIndex", partsPerMap, big, user); } catch (Exception e) { } } }; Thread removeMapThread = new Thread() { @Override public void run() { cache.removeMap("bigIndex"); } }; if (i%2==0) { getInfoThread.start(); removeMapThread.start(); } else { removeMapThread.start(); getInfoThread.start(); } getInfoThread.join(); removeMapThread.join(); assertEquals(true, cache.checkTotalMemoryUsed()); } }