BlockReaderLocalLegacy implements BlockReader { @Override public synchronized void close() throws IOException { IOUtils.cleanup(LOG, dataIn, checksumIn); if (slowReadBuff != null) { bufferPool.returnBuffer(slowReadBuff); slowReadBuff = null; } if (checksumBuff != null) { bufferPool.returnBuffer(checksumBuff); checksumBuff = null; } startOffset = -1; checksum = null; } private  BlockReaderLocalLegacy(DFSClient.Conf conf, String hdfsfile,
      ExtendedBlock block, Token<BlockTokenIdentifier> token, long startOffset,
      long length, BlockLocalPathInfo pathinfo, FileInputStream dataIn); private  BlockReaderLocalLegacy(DFSClient.Conf conf, String hdfsfile,
      ExtendedBlock block, Token<BlockTokenIdentifier> token, long startOffset,
      long length, BlockLocalPathInfo pathinfo, DataChecksum checksum,
      boolean verifyChecksum, FileInputStream dataIn, long firstChunkOffset,
      FileInputStream checksumIn); @Override synchronized int read(ByteBuffer buf); @Override synchronized int read(byte[] buf, int off, int len); @Override synchronized long skip(long n); @Override synchronized void close(); @Override int readAll(byte[] buf, int offset, int len); @Override void readFully(byte[] buf, int off, int len); @Override int available(); @Override boolean isLocal(); @Override boolean isShortCircuit(); @Override ClientMmap getClientMmap(EnumSet<ReadOption> opts);  }
@Test(timeout=20000) public void testBlockReaderLocalLegacyWithAppend() throws Exception { final short REPL_FACTOR = 1; final HdfsConfiguration conf = getConfiguration(null); conf.setBoolean(DFSConfigKeys.DFS_CLIENT_USE_LEGACY_BLOCKREADERLOCAL, true); final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).build(); cluster.waitActive(); final DistributedFileSystem dfs = cluster.getFileSystem(); final Path path = new Path("/testBlockReaderLocalLegacy"); DFSTestUtil.createFile(dfs, path, 10, REPL_FACTOR, 0); DFSTestUtil.waitReplication(dfs, path, REPL_FACTOR); final ClientDatanodeProtocol proxy; final Token<BlockTokenIdentifier> token; final ExtendedBlock originalBlock; final long originalGS; { final LocatedBlock lb = cluster.getNameNode().getRpcServer() .getBlockLocations(path.toString(), 0, 1).get(0); proxy = DFSUtil.createClientDatanodeProtocolProxy( lb.getLocations()[0], conf, 60000, false); token = lb.getBlockToken(); final ExtendedBlock blk = new ExtendedBlock(lb.getBlock()); originalBlock = new ExtendedBlock(blk); originalGS = originalBlock.getGenerationStamp(); final BlockLocalPathInfo info = proxy.getBlockLocalPathInfo(blk, token); Assert.assertEquals(originalGS, info.getBlock().getGenerationStamp()); } { FSDataOutputStream out = dfs.append(path); out.write(1); out.close(); } { final LocatedBlock lb = cluster.getNameNode().getRpcServer() .getBlockLocations(path.toString(), 0, 1).get(0); final long newGS = lb.getBlock().getGenerationStamp(); Assert.assertTrue(newGS > originalGS); Assert.assertEquals(originalGS, originalBlock.getGenerationStamp()); final BlockLocalPathInfo info = proxy.getBlockLocalPathInfo( originalBlock, token); Assert.assertEquals(newGS, info.getBlock().getGenerationStamp()); } cluster.shutdown(); }