HBASE-21047 Object creation of StoreFileScanner thru constructor and close may leave refCount to -1 (Vishal Khandelwal)

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java

@@ -1219,12 +1219,18 @@ public class StoreFile {
      */
     public StoreFileScanner getStoreFileScanner(boolean cacheBlocks, boolean pread,
         boolean isCompaction, long readPt, long scannerOrder, boolean canOptimizeForNonNullColumn) {
-      // Increment the ref count
-      refCount.incrementAndGet();
       return new StoreFileScanner(this, getScanner(cacheBlocks, pread, isCompaction), !isCompaction,
           reader.hasMVCCInfo(), readPt, scannerOrder, canOptimizeForNonNullColumn);
     }
 
+    /**
+     * Increment the ref count associated with the reader when ever a scanner associated with the
+     * reader is opened
+     */
+    void incrementRefCount() {
+      refCount.incrementAndGet();
+    }
+
     /**
      * Decrement the ref count associated with the reader when ever a scanner associated
      * with the reader is closed

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFileScanner.java

@@ -94,6 +94,7 @@ public class StoreFileScanner implements KeyValueScanner {
     this.hasMVCCInfo = hasMVCC;
     this.scannerOrder = scannerOrder;
     this.canOptimizeForNonNullColumn = canOptimizeForNonNullColumn;
+    this.reader.incrementRefCount();
   }
 
   boolean isPrimaryReplica() {

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreFile.java

@@ -234,6 +234,29 @@ public class TestStoreFile extends HBaseTestCase {
     assertEquals((LAST_CHAR - FIRST_CHAR + 1) * (LAST_CHAR - FIRST_CHAR + 1), count);
   }
 
+  @Test
+  public void testStoreFileReference() throws Exception {
+    Path f = new Path(ROOT_DIR, getName());
+    HFileContext meta = new HFileContextBuilder().withBlockSize(8 * 1024).build();
+    // Make a store file and write data to it.
+    StoreFile.Writer writer = new StoreFile.WriterBuilder(conf, cacheConf, this.fs).withFilePath(f)
+        .withFileContext(meta).build();
+
+    writeStoreFile(writer);
+    writer.close();
+
+    // Creates a reader for StoreFile
+    StoreFile.Reader reader = new StoreFile.Reader(this.fs, f, cacheConf, conf);
+    StoreFileScanner scanner =
+        new StoreFileScanner(reader, mock(HFileScanner.class), false, false, 0, 0, false);
+
+    // Verify after instantiating scanner refCount is increased
+    assertTrue(scanner.getReader().isReferencedInReads());
+    scanner.close();
+    // Verify after closing scanner refCount is decreased
+    assertFalse(scanner.getReader().isReferencedInReads());
+  }
+
   @Test
   public void testEmptyStoreFileRestrictKeyRanges() throws Exception {
     StoreFile.Reader reader = mock(StoreFile.Reader.class);