@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();
}