BlockReaderLocalLegacy implements BlockReader { @Override public synchronized int read(ByteBuffer buf) throws IOException { int nRead = 0; if (verifyChecksum) { if (slowReadBuff.hasRemaining()) { int fromSlowReadBuff = Math.min(buf.remaining(), slowReadBuff.remaining()); writeSlice(slowReadBuff, buf, fromSlowReadBuff); nRead += fromSlowReadBuff; } if (buf.remaining() >= bytesPerChecksum && offsetFromChunkBoundary == 0) { int len = buf.remaining() - (buf.remaining() % bytesPerChecksum); len = Math.min(len, slowReadBuff.capacity()); int oldlimit = buf.limit(); buf.limit(buf.position() + len); int readResult = 0; try { readResult = doByteBufferRead(buf); } finally { buf.limit(oldlimit); } if (readResult == -1) { return nRead; } else { nRead += readResult; buf.position(buf.position() + readResult); } } if ((buf.remaining() > 0 && buf.remaining() < bytesPerChecksum) || offsetFromChunkBoundary > 0) { int toRead = Math.min(buf.remaining(), bytesPerChecksum - offsetFromChunkBoundary); int readResult = fillSlowReadBuffer(toRead); if (readResult == -1) { return nRead; } else { int fromSlowReadBuff = Math.min(readResult, buf.remaining()); writeSlice(slowReadBuff, buf, fromSlowReadBuff); nRead += fromSlowReadBuff; } } } else { nRead = doByteBufferRead(buf); if (nRead > 0) { buf.position(buf.position() + nRead); } } return nRead; } private  BlockReaderLocalLegacy(ShortCircuitConf conf, String hdfsfile,
      ExtendedBlock block, Token<BlockTokenIdentifier> token, long startOffset,
      long length, BlockLocalPathInfo pathinfo, FileInputStream dataIn, Tracer tracer); private  BlockReaderLocalLegacy(ShortCircuitConf conf, String hdfsfile,
      ExtendedBlock block, Token<BlockTokenIdentifier> token, long startOffset,
      long length, BlockLocalPathInfo pathinfo, DataChecksum checksum,
      boolean verifyChecksum, FileInputStream dataIn, long firstChunkOffset,
      FileInputStream checksumIn, Tracer tracer); @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 public void testStablePositionAfterCorruptRead() throws Exception { final short REPL_FACTOR = 1; final long FILE_LENGTH = 512L; HdfsConfiguration conf = getConfiguration(null); MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).build(); cluster.waitActive(); FileSystem fs = cluster.getFileSystem(); Path path = new Path("/corrupted"); DFSTestUtil.createFile(fs, path, FILE_LENGTH, REPL_FACTOR, 12345L); DFSTestUtil.waitReplication(fs, path, REPL_FACTOR); ExtendedBlock block = DFSTestUtil.getFirstBlock(fs, path); int blockFilesCorrupted = cluster.corruptBlockOnDataNodes(block); assertEquals("All replicas not corrupted", REPL_FACTOR, blockFilesCorrupted); FSDataInputStream dis = cluster.getFileSystem().open(path); ByteBuffer buf = ByteBuffer.allocateDirect((int)FILE_LENGTH); boolean sawException = false; try { dis.read(buf); } catch (ChecksumException ex) { sawException = true; } assertTrue(sawException); assertEquals(0, buf.position()); assertEquals(buf.capacity(), buf.limit()); dis = cluster.getFileSystem().open(path); buf.position(3); buf.limit(25); sawException = false; try { dis.read(buf); } catch (ChecksumException ex) { sawException = true; } assertTrue(sawException); assertEquals(3, buf.position()); assertEquals(25, buf.limit()); cluster.shutdown(); }