@Test(timeout=60000)
public void testShmBasedStaleness() throws Exception {

  BlockReaderTestUtil.enableShortCircuitShmTracing();
  TemporarySocketDirectory sockDir = new TemporarySocketDirectory();
  Configuration conf = createShortCircuitConf("testShmBasedStaleness", sockDir);
  MiniDFSCluster cluster =
      new MiniDFSCluster.Builder(conf).numDataNodes(1).build();
  cluster.waitActive();
  DistributedFileSystem fs = cluster.getFileSystem();
  final ShortCircuitCache cache =
      fs.getClient().getClientContext().getShortCircuitCache();
  String TEST_FILE = "/test_file";
  final int TEST_FILE_LEN = 8193;
  final int SEED = 0xFADED;
  DFSTestUtil.createFile(fs, new Path(TEST_FILE), TEST_FILE_LEN,
      (short)1, SEED);
  FSDataInputStream fis = fs.open(new Path(TEST_FILE));
  int first = fis.read();
  final ExtendedBlock block =
      DFSTestUtil.getFirstBlock(fs, new Path(TEST_FILE));
  Assert.assertTrue(first != -1);
  cache.accept(new CacheVisitor() {
    @Override
    public void visit(int numOutstandingMmaps,
        Map<ExtendedBlockId, ShortCircuitReplica> replicas,
        Map<ExtendedBlockId, InvalidToken> failedLoads,
        Map<Long, ShortCircuitReplica> evictable,
        Map<Long, ShortCircuitReplica> evictableMmapped) {
      ShortCircuitReplica replica = replicas.get(
          ExtendedBlockId.fromExtendedBlock(block));
      Assert.assertNotNull(replica);
      Assert.assertTrue(replica.getSlot().isValid());
    }
  });
  // Stop the Namenode.  This will close the socket keeping the client's
  // shared memory segment alive, and make it stale.
  cluster.getDataNodes().get(0).shutdown();
  cache.accept(new CacheVisitor() {
    @Override
    public void visit(int numOutstandingMmaps,
        Map<ExtendedBlockId, ShortCircuitReplica> replicas,
        Map<ExtendedBlockId, InvalidToken> failedLoads,
        Map<Long, ShortCircuitReplica> evictable,
        Map<Long, ShortCircuitReplica> evictableMmapped) {
      ShortCircuitReplica replica = replicas.get(
          ExtendedBlockId.fromExtendedBlock(block));
      Assert.assertNotNull(replica);
      Assert.assertFalse(replica.getSlot().isValid());
    }
  });
  cluster.shutdown();
}