From f6fc94ede999766d37f38828fe36b581924fcc30 Mon Sep 17 00:00:00 2001
From: tedyu <yuzhihong@gmail.com>
Date: Thu, 17 Nov 2016 07:52:52 -0800
Subject: [PATCH] HBASE-17118 StoreScanner leaked in KeyValueHeap (binlijin)

---
 .../hbase/regionserver/KeyValueHeap.java      | 72 +++++++++++--------
 1 file changed, 44 insertions(+), 28 deletions(-)

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueHeap.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueHeap.java
index 9ece14b968f..f2b453a4b02 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueHeap.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueHeap.java
@@ -26,6 +26,8 @@ import java.util.List;
 import java.util.PriorityQueue;
 import java.util.Set;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellComparator;
 import org.apache.hadoop.hbase.classification.InterfaceAudience;
@@ -46,6 +48,7 @@ import org.apache.hadoop.hbase.regionserver.ScannerContext.NextState;
 @InterfaceAudience.Private
 public class KeyValueHeap extends NonReversedNonLazyKeyValueScanner
     implements KeyValueScanner, InternalScanner {
+  private static final Log LOG = LogFactory.getLog(KeyValueHeap.class);
   protected PriorityQueue<KeyValueScanner> heap = null;
   // Holds the scanners when a ever a eager close() happens.  All such eagerly closed
   // scans are collected and when the final scanner.close() happens will perform the
@@ -292,38 +295,51 @@ public class KeyValueHeap extends NonReversedNonLazyKeyValueScanner
     }
 
     KeyValueScanner scanner = current;
-    while (scanner != null) {
-      Cell topKey = scanner.peek();
-      if (comparator.getComparator().compare(seekKey, topKey) <= 0) {
-        // Top KeyValue is at-or-after Seek KeyValue. We only know that all
-        // scanners are at or after seekKey (because fake keys of
-        // scanners where a lazy-seek operation has been done are not greater
-        // than their real next keys) but we still need to enforce our
-        // invariant that the top scanner has done a real seek. This way
-        // StoreScanner and RegionScanner do not have to worry about fake keys.
-        heap.add(scanner);
-        current = pollRealKV();
-        return current != null;
-      }
+    try {
+      while (scanner != null) {
+        Cell topKey = scanner.peek();
+        if (comparator.getComparator().compare(seekKey, topKey) <= 0) {
+          // Top KeyValue is at-or-after Seek KeyValue. We only know that all
+          // scanners are at or after seekKey (because fake keys of
+          // scanners where a lazy-seek operation has been done are not greater
+          // than their real next keys) but we still need to enforce our
+          // invariant that the top scanner has done a real seek. This way
+          // StoreScanner and RegionScanner do not have to worry about fake
+          // keys.
+          heap.add(scanner);
+          scanner = null;
+          current = pollRealKV();
+          return current != null;
+        }
 
-      boolean seekResult;
-      if (isLazy && heap.size() > 0) {
-        // If there is only one scanner left, we don't do lazy seek.
-        seekResult = scanner.requestSeek(seekKey, forward, useBloom);
-      } else {
-        seekResult = NonLazyKeyValueScanner.doRealSeek(
-            scanner, seekKey, forward);
-      }
+        boolean seekResult;
+        if (isLazy && heap.size() > 0) {
+          // If there is only one scanner left, we don't do lazy seek.
+          seekResult = scanner.requestSeek(seekKey, forward, useBloom);
+        } else {
+          seekResult = NonLazyKeyValueScanner.doRealSeek(scanner, seekKey,
+              forward);
+        }
 
-      if (!seekResult) {
-        this.scannersForDelayedClose.add(scanner);
-      } else {
-        heap.add(scanner);
+        if (!seekResult) {
+          this.scannersForDelayedClose.add(scanner);
+        } else {
+          heap.add(scanner);
+        }
+        scanner = heap.poll();
+        if (scanner == null) {
+          current = null;
+        }
       }
-      scanner = heap.poll();
-      if (scanner == null) {
-        current = null;
+    } catch (Exception e) {
+      if (scanner != null) {
+        try {
+          scanner.close();
+        } catch (Exception ce) {
+          LOG.warn("close KeyValueScanner error", ce);
+        }
       }
+      throw e;
     }
 
     // Heap is returning empty, scanner is done