HDFS-7744. Fix potential NPE in DFSInputStream after setDropBehind or setReadahead is called (cmccabe)

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

@@ -890,6 +890,9 @@ Release 2.7.0 - UNRELEASED
     HDFS-7756. Restore method signature for LocatedBlock#getLocations(). (Ted
     Yu via yliu)
 
+    HDFS-7744. Fix potential NPE in DFSInputStream after setDropBehind or
+    setReadahead is called (cmccabe)
+
 Release 2.6.1 - UNRELEASED
 
   INCOMPATIBLE CHANGES

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

@@ -580,10 +580,7 @@ implements ByteBufferReadable, CanSetDropBehind, CanSetReadahead,
     }
 
     // Will be getting a new BlockReader.
-    if (blockReader != null) {
+    closeCurrentBlockReader();
-      blockReader.close();
-      blockReader = null;
-    }
 
     //
     // Connect to best DataNode for desired Block, with potential offset
@@ -686,10 +683,7 @@ implements ByteBufferReadable, CanSetDropBehind, CanSetReadahead,
           "unreleased ByteBuffers allocated by read().  " +
           "Please release " + builder.toString() + ".");
     }
-    if (blockReader != null) {
+    closeCurrentBlockReader();
-      blockReader.close();
-      blockReader = null;
-    }
     super.close();
   }
 
@@ -1649,6 +1643,7 @@ implements ByteBufferReadable, CanSetDropBehind, CanSetReadahead,
       DFSClient.LOG.error("error closing blockReader", e);
     }
     blockReader = null;
+    blockEnd = -1;
   }
 
   @Override

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestCachingStrategy.java

@@ -369,4 +369,34 @@ public class TestCachingStrategy {
       }
     }
   }
+
+  @Test(timeout=120000)
+  public void testSeekAfterSetDropBehind() throws Exception {
+    // start a cluster
+    LOG.info("testSeekAfterSetDropBehind");
+    Configuration conf = new HdfsConfiguration();
+    MiniDFSCluster cluster = null;
+    String TEST_PATH = "/test";
+    int TEST_PATH_LEN = MAX_TEST_FILE_LEN;
+    try {
+      cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1)
+          .build();
+      cluster.waitActive();
+      FileSystem fs = cluster.getFileSystem();
+      createHdfsFile(fs, new Path(TEST_PATH), TEST_PATH_LEN, false);
+      // verify that we can seek after setDropBehind
+      FSDataInputStream fis = fs.open(new Path(TEST_PATH));
+      try {
+        Assert.assertTrue(fis.read() != -1); // create BlockReader
+        fis.setDropBehind(false); // clear BlockReader
+        fis.seek(2); // seek
+      } finally {
+        fis.close();
+      }
+    } finally {
+      if (cluster != null) {
+        cluster.shutdown();
+      }
+    }
+  }
 }