From 47e12fb3a08d5d81ebcfae1abeb1bea909f76e49 Mon Sep 17 00:00:00 2001
From: Ramkrishna <ramkrishna.s.vasudevan@intel.com>
Date: Wed, 28 Sep 2016 15:39:08 +0530
Subject: [PATCH] HBASE-16696 After HBASE-16604 - does not release blocks in
 case of scanner exception (Ram)

---
 .../hbase/regionserver/RSRpcServices.java     | 25 ++++++++++++-------
 .../client/TestBlockEvictionFromClient.java   | 21 +++-------------
 2 files changed, 20 insertions(+), 26 deletions(-)

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java
index 541680c95fd..c01ea39c50d 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java
@@ -2694,6 +2694,7 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
             rsh.incNextCallSeq();
           }
         }
+        boolean scannerClosed = false;
         try {
           // Remove lease while its being processed in server; protects against case
           // where processing of request takes > lease expiration time.
@@ -2907,6 +2908,8 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
           // fail this RPC and close the scanner while opening up another one from the start of
           // row that the client has last seen.
           closeScanner(region, scanner, scannerName, context);
+          // scanner is closed here
+          scannerClosed = true;
 
           // We closed the scanner already. Instead of throwing the IOException, and client
           // retrying with the same scannerId only to get USE on the next RPC, we directly throw
@@ -2920,15 +2923,19 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
                 + " scanner state for clients older than 1.3.", e);
           }
         } finally {
-          if (context != null) {
-            context.setCallBack(rsh.shippedCallback);
-          }
-          // Adding resets expiration time on lease.
-          if (scanners.containsKey(scannerName)) {
-            ttl = this.scannerLeaseTimeoutPeriod;
-            // When context != null, adding back the lease will be done in callback set above.
-            if (context == null) {
-              if (lease != null) regionServer.leases.addLease(lease);
+          // If the scanner is not closed, set the shipped callback
+          if (!scannerClosed) {
+            if (context != null) {
+              context.setCallBack(rsh.shippedCallback);
+            }
+
+            // Adding resets expiration time on lease.
+            if (scanners.containsKey(scannerName)) {
+              ttl = this.scannerLeaseTimeoutPeriod;
+              // When context != null, adding back the lease will be done in callback set above.
+              if (context == null) {
+                if (lease != null) regionServer.leases.addLease(lease);
+              }
             }
           }
         }
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestBlockEvictionFromClient.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestBlockEvictionFromClient.java
index ba75d6e1465..3b19bcc72d9 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestBlockEvictionFromClient.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestBlockEvictionFromClient.java
@@ -18,6 +18,7 @@
 package org.apache.hadoop.hbase.client;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
 import java.io.IOException;
@@ -1185,23 +1186,9 @@ public class TestBlockEvictionFromClient {
           usedBlocksFound = true;
         }
       }
-      assertTrue(usedBlocksFound);
-      // Sleep till the scan lease would expire? Can we reduce this value?
-      Thread.sleep(5100);
-      iterator = cache.iterator();
-      refCount = 0;
-      while (iterator.hasNext()) {
-        CachedBlock next = iterator.next();
-        BlockCacheKey cacheKey = new BlockCacheKey(next.getFilename(), next.getOffset());
-        if (cache instanceof BucketCache) {
-          refCount = ((BucketCache) cache).getRefCount(cacheKey);
-        } else if (cache instanceof CombinedBlockCache) {
-          refCount = ((CombinedBlockCache) cache).getRefCount(cacheKey);
-        } else {
-          continue;
-        }
-        assertEquals(0, refCount);
-      }
+      assertFalse(usedBlocksFound);
+      // you should always see 0 ref count. since after HBASE-16604 we always recreate the scanner
+      assertEquals(0, refCount);
     } finally {
       if (table != null) {
         table.close();