public long getNumBlocksFailedToCache() {
    return numBlocksFailedToCache.longValue();
  }
@Test(timeout=600000)
  public void testFilesExceedMaxLockedMemory() throws Exception {
    LOG.info("beginning testFilesExceedMaxLockedMemory");

    // Create some test files that will exceed total cache capacity
    final int numFiles = 5;
    final long fileSize = CACHE_CAPACITY / (numFiles-1);

    final Path[] testFiles = new Path[numFiles];
    final HdfsBlockLocation[][] fileLocs = new HdfsBlockLocation[numFiles][];
    final long[] fileSizes = new long[numFiles];
    for (int i=0; i<numFiles; i++) {
      testFiles[i] = new Path("/testFilesExceedMaxLockedMemory-" + i);
      DFSTestUtil.createFile(fs, testFiles[i], fileSize, (short)1, 0xDFAl);
      fileLocs[i] = (HdfsBlockLocation[])fs.getFileBlockLocations(
          testFiles[i], 0, fileSize);
      // Get the file size (sum of blocks)
      long[] sizes = getBlockSizes(fileLocs[i]);
      for (int j=0; j<sizes.length; j++) {
        fileSizes[i] += sizes[j];
      }
    }

    // Cache the first n-1 files
    long total = 0;
    DFSTestUtil.verifyExpectedCacheUsage(0, 0, fsd);
    for (int i=0; i<numFiles-1; i++) {
      setHeartbeatResponse(cacheBlocks(fileLocs[i]));
      total = DFSTestUtil.verifyExpectedCacheUsage(
          rounder.roundUp(total + fileSizes[i]), 4 * (i + 1), fsd);
    }

    // nth file should hit a capacity exception
    final LogVerificationAppender appender = new LogVerificationAppender();
    final Logger logger = Logger.getRootLogger();
    logger.addAppender(appender);
    setHeartbeatResponse(cacheBlocks(fileLocs[numFiles-1]));

    GenericTestUtils.waitFor(new Supplier<Boolean>() {
      @Override
      public Boolean get() {
        // check the log reported by FsDataSetCache
        // in the case that cache capacity is exceeded.
        int lines = appender.countLinesWithMessage(
            "could not reserve more bytes in the cache: ");
        return lines > 0;
      }
    }, 500, 30000);
    // Also check the metrics for the failure
    assertTrue("Expected more than 0 failed cache attempts",
        fsd.getNumBlocksFailedToCache() > 0);

    // Uncache the n-1 files
    int curCachedBlocks = 16;
    for (int i=0; i<numFiles-1; i++) {
      setHeartbeatResponse(uncacheBlocks(fileLocs[i]));
      long uncachedBytes = rounder.roundUp(fileSizes[i]);
      total -= uncachedBytes;
      curCachedBlocks -= uncachedBytes / BLOCK_SIZE;
      DFSTestUtil.verifyExpectedCacheUsage(total, curCachedBlocks, fsd);
    }
    LOG.info("finishing testFilesExceedMaxLockedMemory");
  }