From 471f942ec8a11bb5de022f8a05af93e9c0082457 Mon Sep 17 00:00:00 2001
From: Yu Li <liyu@apache.org>
Date: Tue, 21 Jun 2016 20:06:33 +0800
Subject: [PATCH] HBASE-16032 Possible memory leak in StoreScanner

---
 .../hadoop/hbase/regionserver/HRegion.java    | 41 ++++++++++++-------
 .../hbase/regionserver/StoreScanner.java      | 37 ++++++++++-------
 2 files changed, 49 insertions(+), 29 deletions(-)

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
index 4b4ba9b01a1..bf7f31cc84b 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
@@ -5630,26 +5630,39 @@ 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);
       }
 
-      for (Map.Entry<byte[], NavigableSet<byte[]>> entry : scan.getFamilyMap().entrySet()) {
-        Store store = stores.get(entry.getKey());
-        KeyValueScanner scanner;
-        try {
-          scanner = store.getScanner(scan, entry.getValue(), this.readPt);
-        } catch (FileNotFoundException e) {
-          throw handleFileNotFound(e);
+      try {
+        for (Map.Entry<byte[], NavigableSet<byte[]>> entry : scan.getFamilyMap().entrySet()) {
+          Store store = stores.get(entry.getKey());
+          KeyValueScanner scanner;
+          try {
+            scanner = store.getScanner(scan, entry.getValue(), this.readPt);
+          } catch (FileNotFoundException e) {
+            throw handleFileNotFound(e);
+          }
+          instantiatedScanners.add(scanner);
+          if (this.filter == null || !scan.doLoadColumnFamiliesOnDemand()
+              || this.filter.isFamilyEssential(entry.getKey())) {
+            scanners.add(scanner);
+          } else {
+            joinedScanners.add(scanner);
+          }
         }
-        if (this.filter == null || !scan.doLoadColumnFamiliesOnDemand()
-          || this.filter.isFamilyEssential(entry.getKey())) {
-          scanners.add(scanner);
-        } else {
-          joinedScanners.add(scanner);
+        initializeKVHeap(scanners, joinedScanners, region);
+      } catch (IOException e) {
+        // close all already instantiated scanners before throwing the exception
+        for (KeyValueScanner scanner : instantiatedScanners) {
+          scanner.close();
         }
+        throw e;
       }
-      initializeKVHeap(scanners, joinedScanners, region);
     }
 
     protected void initializeKVHeap(List<KeyValueScanner> scanners,
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
index 7ebae94bd20..ceb4b8e9bdf 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
@@ -202,24 +202,31 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
 
     this.store.addChangedReaderObserver(this);
 
-    // Pass columns to try to filter out unnecessary StoreFiles.
-    List<KeyValueScanner> scanners = getScannersNoCompaction();
+    try {
+      // Pass columns to try to filter out unnecessary StoreFiles.
+      List<KeyValueScanner> scanners = getScannersNoCompaction();
 
-    // Seek all scanners to the start of the Row (or if the exact matching row
-    // 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
-        && lazySeekEnabledGlobally, parallelSeekEnabled);
+      // Seek all scanners to the start of the Row (or if the exact matching row
+      // 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 && lazySeekEnabledGlobally,
+        parallelSeekEnabled);
 
-    // set storeLimit
-    this.storeLimit = scan.getMaxResultsPerColumnFamily();
+      // set storeLimit
+      this.storeLimit = scan.getMaxResultsPerColumnFamily();
 
-    // set rowOffset
-    this.storeOffset = scan.getRowOffsetPerColumnFamily();
-    addCurrentScanners(scanners);
-    // Combine all seeked scanners with a heap
-    resetKVHeap(scanners, store.getComparator());
+      // set rowOffset
+      this.storeOffset = scan.getRowOffsetPerColumnFamily();
+      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;
+    }
   }
 
   /**