HBASE-20403. Fix race between prefetch task and non-pread HFile reads

With prefetch-on-open enabled, the task doing the prefetching was using
non-positional (i.e. streaming) reads. If the main (non-prefetch) thread
was also using non-positional reads, these two would conflict, because
inputstreams are not thread-safe for non-positional reads.

In the case of an encrypted filesystem, this could cause JVM crashes,
etc, as underlying cipher buffers were freed underneath the racing
threads. In the case of a non-encrypted filesystem, less severe errors
would be thrown. The included unit test reproduces the latter case.

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileReaderImpl.java

@@ -275,8 +275,8 @@ public class HFileReaderImpl implements HFile.Reader, Configurable {
               // next header, will not have happened...so, pass in the onDiskSize gotten from the
               // cached block. This 'optimization' triggers extremely rarely I'd say.
               long onDiskSize = prevBlock != null? prevBlock.getNextBlockOnDiskSize(): -1;
-              HFileBlock block = readBlock(offset, onDiskSize, true, false, false, false,
+              HFileBlock block = readBlock(offset, onDiskSize, /*cacheBlock=*/true,
-                  null, null);
+                  /*pread=*/true, false, false, null, null);
               // Need not update the current block. Ideally here the readBlock won't find the
               // block in cache. We call this readBlock so that block data is read from FS and
               // cached in BC. So there is no reference count increment that happens here.

hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestPrefetch.java

@@ -79,10 +79,35 @@ public class TestPrefetch {
 
   @Test
   public void testPrefetch() throws Exception {
-    Path storeFile = writeStoreFile();
+    Path storeFile = writeStoreFile("TestPrefetch");
     readStoreFile(storeFile);
   }
 
+  @Test
+  public void testPrefetchRace() throws Exception {
+    for (int i = 0; i < 10; i++) {
+      Path storeFile = writeStoreFile("TestPrefetchRace-" + i);
+      readStoreFileLikeScanner(storeFile);
+    }
+  }
+
+  /**
+   * Read a storefile in the same manner as a scanner -- using non-positional reads and
+   * without waiting for prefetch to complete.
+   */
+  private void readStoreFileLikeScanner(Path storeFilePath) throws Exception {
+    // Open the file
+    HFile.Reader reader = HFile.createReader(fs, storeFilePath, cacheConf, true, conf);
+    do {
+      long offset = 0;
+      while (offset < reader.getTrailer().getLoadOnOpenDataOffset()) {
+        HFileBlock block = reader.readBlock(offset, -1, false, /*pread=*/false,
+            false, true, null, null);
+        offset += block.getOnDiskSizeWithHeader();
+      }
+    } while (!reader.prefetchComplete());
+  }
+
   private void readStoreFile(Path storeFilePath) throws Exception {
     // Open the file
     HFile.Reader reader = HFile.createReader(fs, storeFilePath, cacheConf, true, conf);
@@ -108,8 +133,8 @@ public class TestPrefetch {
     }
   }
 
-  private Path writeStoreFile() throws IOException {
+  private Path writeStoreFile(String fname) throws IOException {
-    Path storeFileParentDir = new Path(TEST_UTIL.getDataTestDir(), "TestPrefetch");
+    Path storeFileParentDir = new Path(TEST_UTIL.getDataTestDir(), fname);
     HFileContext meta = new HFileContextBuilder()
       .withBlockSize(DATA_BLOCK_SIZE)
       .build();