svn merge -c 1565310 merging from trunk to branch-2 to fix: HDFS-5881. Fix skip() of the short-circuit local reader(legacy).

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1565313 13f79535-47bb-0310-9956-ffa450edef68

hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt

@@ -51,6 +51,8 @@ Release 2.4.0 - UNRELEASED
     HDFS-5709. Improve NameNode upgrade with existing reserved paths and path
     components. (Andrew Wang via atm)
 
+    HDFS-5881. Fix skip() of the short-circuit local reader(legacy). (kihwal)
+
 Release 2.3.0 - UNRELEASED
 
   INCOMPATIBLE CHANGES

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/BlockReaderLocalLegacy.java

@@ -629,7 +629,7 @@ class BlockReaderLocalLegacy implements BlockReader {
         skipBuf = new byte[bytesPerChecksum];
       }
       int ret = read(skipBuf, 0, (int)(n - remaining));
-      return ret;
+      return (remaining + ret);
     }
   
     // optimize for big gap: discard the current buffer, skip to
@@ -660,9 +660,9 @@ class BlockReaderLocalLegacy implements BlockReader {
     int ret = read(skipBuf, 0, myOffsetFromChunkBoundary);
 
     if (ret == -1) {  // EOS
-      return toskip;
+      return (toskip + remaining);
     } else {
-      return (toskip + ret);
+      return (toskip + remaining + ret);
     }
   }
 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java

@@ -1345,6 +1345,14 @@ implements ByteBufferReadable, CanSetDropBehind, CanSetReadahead,
           pos += blockReader.skip(diff);
           if (pos == targetPos) {
             done = true;
+          } else {
+            // The range was already checked. If the block reader returns
+            // something unexpected instead of throwing an exception, it is
+            // most likely a bug. 
+            String errMsg = "BlockReader failed to seek to " + 
+                targetPos + ". Instead, it seeked to " + pos + ".";
+            DFSClient.LOG.warn(errMsg);
+            throw new IOException(errMsg);
           }
         } catch (IOException e) {//make following read to retry
           if(DFSClient.LOG.isDebugEnabled()) {