From ee5a29e1f3cfff1ded68a232175812c4c0ca28f3 Mon Sep 17 00:00:00 2001
From: "Ma,Gang" <ganma@ebay.com>
Date: Tue, 21 Jun 2016 09:57:12 +0800
Subject: [PATCH] KYLIN-1740 HDFS storage leak in
 HBaseResourceStore.deleteResourceImpl() in case of large cell of dictionary

---
 .../kylin/storage/hbase/HBaseResourceStore.java    | 41 +++++++++++++++++-----
 1 file changed, 32 insertions(+), 9 deletions(-)

diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseResourceStore.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseResourceStore.java
index a389240..c0367bf 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseResourceStore.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseResourceStore.java
@@ -267,9 +267,28 @@ public class HBaseResourceStore extends ResourceStore {
     protected void deleteResourceImpl(String resPath) throws IOException {
         HTableInterface table = getConnection().getTable(getAllInOneTableName());
         try {
+            boolean hdfsResourceExist = false;
+            Result result = internalGetFromHTable(table,resPath,true,false);
+            if (result != null) {
+                byte[] value = result.getValue(B_FAMILY, B_COLUMN);
+                if (value != null && value.length == 0) {
+                    hdfsResourceExist = true;
+                }
+            }
+
             Delete del = new Delete(Bytes.toBytes(resPath));
             table.delete(del);
             table.flushCommits();
+
+            if (hdfsResourceExist) { // remove hdfs cell value
+                Path redirectPath = bigCellHDFSPath(resPath);
+                Configuration hconf = HBaseConnection.getCurrentHBaseConfiguration();
+                FileSystem fileSystem = FileSystem.get(hconf);
+
+                if (fileSystem.exists(redirectPath)) {
+                    fileSystem.delete(redirectPath, true);
+                }
+            }
         } finally {
             IOUtils.closeQuietly(table);
         }
@@ -281,10 +300,19 @@ public class HBaseResourceStore extends ResourceStore {
     }
 
     private Result getFromHTable(String path, boolean fetchContent, boolean fetchTimestamp) throws IOException {
+        HTableInterface table = getConnection().getTable(getAllInOneTableName());
+        try {
+            return internalGetFromHTable(table,path,fetchContent,fetchTimestamp);
+        } finally {
+            IOUtils.closeQuietly(table);
+        }
+    }
+
+    private Result internalGetFromHTable(HTableInterface table, String path, boolean fetchContent, boolean fetchTimestamp) throws IOException {
         byte[] rowkey = Bytes.toBytes(path);
 
         Get get = new Get(rowkey);
-        
+
         if (!fetchContent && !fetchTimestamp) {
             get.setCheckExistenceOnly(true);
         } else {
@@ -294,14 +322,9 @@ public class HBaseResourceStore extends ResourceStore {
                 get.addColumn(B_FAMILY, B_COLUMN_TS);
         }
 
-        HTableInterface table = getConnection().getTable(getAllInOneTableName());
-        try {
-            Result result = table.get(get);
-            boolean exists = result != null && (!result.isEmpty() || (result.getExists() != null && result.getExists()));
-            return exists ? result : null;
-        } finally {
-            IOUtils.closeQuietly(table);
-        }
+        Result result = table.get(get);
+        boolean exists = result != null && (!result.isEmpty() || (result.getExists() != null && result.getExists()));
+        return exists ? result : null;
     }
 
     private byte[] plusZero(byte[] startRow) {
-- 
2.6.4