HBASE-16032 Possible memory leak in StoreScanner

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

@@ -5688,10 +5688,15 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
       List<KeyValueScanner> scanners = new ArrayList<KeyValueScanner>(scan.getFamilyMap().size());
       List<KeyValueScanner> joinedScanners
         = new ArrayList<KeyValueScanner>(scan.getFamilyMap().size());
-      if (additionalScanners != null) {
+      // Store all already instantiated scanners for exception handling
+      List<KeyValueScanner> instantiatedScanners = new ArrayList<KeyValueScanner>();
+      // handle additionalScanners
+      if (additionalScanners != null && !additionalScanners.isEmpty()) {
         scanners.addAll(additionalScanners);
+        instantiatedScanners.addAll(additionalScanners);
       }
 
+      try {
         for (Map.Entry<byte[], NavigableSet<byte[]>> entry : scan.getFamilyMap().entrySet()) {
           Store store = stores.get(entry.getKey());
           KeyValueScanner scanner;
@@ -5700,6 +5705,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
           } catch (FileNotFoundException e) {
             throw handleFileNotFound(e);
           }
+          instantiatedScanners.add(scanner);
           if (this.filter == null || !scan.doLoadColumnFamiliesOnDemand()
               || this.filter.isFamilyEssential(entry.getKey())) {
             scanners.add(scanner);
@@ -5708,6 +5714,13 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
           }
         }
         initializeKVHeap(scanners, joinedScanners, region);
+      } catch (IOException e) {
+        // close all already instantiated scanners before throwing the exception
+        for (KeyValueScanner scanner : instantiatedScanners) {
+          scanner.close();
+        }
+        throw e;
+      }
     }
 
     protected void initializeKVHeap(List<KeyValueScanner> scanners,

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

@@ -198,6 +198,7 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
 
     this.store.addChangedReaderObserver(this);
 
+    try {
       // Pass columns to try to filter out unnecessary StoreFiles.
       List<KeyValueScanner> scanners = getScannersNoCompaction();
 
@@ -205,8 +206,8 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
       // key does not exist, then to the start of the next matching Row).
       // Always check bloom filter to optimize the top row seek for delete
       // family marker.
-    seekScanners(scanners, matcher.getStartKey(), explicitColumnQuery
+      seekScanners(scanners, matcher.getStartKey(), explicitColumnQuery && lazySeekEnabledGlobally,
-        && lazySeekEnabledGlobally, parallelSeekEnabled);
+        parallelSeekEnabled);
 
       // set storeLimit
       this.storeLimit = scan.getMaxResultsPerColumnFamily();
@@ -216,6 +217,12 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
       addCurrentScanners(scanners);
       // Combine all seeked scanners with a heap
       resetKVHeap(scanners, store.getComparator());
+    } catch (IOException e) {
+      // remove us from the HStore#changedReaderObservers here or we'll have no chance to
+      // and might cause memory leak
+      this.store.deleteChangedReaderObserver(this);
+      throw e;
+    }
   }
 
   /**