Repository: hive
Updated Branches:
  refs/heads/branch-2.0 f4468ce68 -> eda730320


HIVE-13151 : Clean up UGI objects in FileSystem cache for transactions (Wei 
Zheng, reviewed by Eugene Koifman)


Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/1785ca00
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/1785ca00
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/1785ca00

Branch: refs/heads/branch-2.0
Commit: 1785ca000596177c28511dc151deb967c3ce1710
Parents: f4468ce
Author: Wei Zheng <w...@apache.org>
Authored: Thu Mar 24 17:29:59 2016 -0700
Committer: Wei Zheng <w...@apache.org>
Committed: Wed Mar 30 15:10:42 2016 -0700

----------------------------------------------------------------------
 .../hive/hcatalog/streaming/HiveEndPoint.java   | 11 +++++
 .../hadoop/hive/ql/txn/compactor/Cleaner.java   |  5 +++
 .../hive/ql/txn/compactor/CompactorThread.java  |  5 +++
 .../hadoop/hive/ql/txn/compactor/Initiator.java |  9 +++-
 .../hadoop/hive/ql/txn/compactor/Worker.java    |  6 +++
 .../apache/hadoop/hive/ql/TestTxnCommands2.java | 47 ++++++++++++++++++++
 6 files changed, 82 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hive/blob/1785ca00/hcatalog/streaming/src/java/org/apache/hive/hcatalog/streaming/HiveEndPoint.java
----------------------------------------------------------------------
diff --git 
a/hcatalog/streaming/src/java/org/apache/hive/hcatalog/streaming/HiveEndPoint.java
 
b/hcatalog/streaming/src/java/org/apache/hive/hcatalog/streaming/HiveEndPoint.java
index 4c77842..baeafad 100644
--- 
a/hcatalog/streaming/src/java/org/apache/hive/hcatalog/streaming/HiveEndPoint.java
+++ 
b/hcatalog/streaming/src/java/org/apache/hive/hcatalog/streaming/HiveEndPoint.java
@@ -18,6 +18,7 @@
 
 package org.apache.hive.hcatalog.streaming;
 
+import org.apache.hadoop.fs.FileSystem;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.hadoop.hive.cli.CliSessionState;
@@ -342,6 +343,11 @@ public class HiveEndPoint {
                 return null;
               }
             } );
+        try {
+          FileSystem.closeAllForUGI(ugi);
+        } catch (IOException exception) {
+          LOG.error("Could not clean up file-system handles for UGI: " + ugi, 
exception);
+        }
       } catch (IOException e) {
         LOG.error("Error closing connection to " + endPt, e);
       } catch (InterruptedException e) {
@@ -937,6 +943,11 @@ public class HiveEndPoint {
                   }
                 }
         );
+        try {
+          FileSystem.closeAllForUGI(ugi);
+        } catch (IOException exception) {
+          LOG.error("Could not clean up file-system handles for UGI: " + ugi, 
exception);
+        }
       } catch (IOException e) {
         throw new ImpersonationFailed("Failed closing Txn Batch as user '" + 
username +
                 "' on  endPoint :" + endPt, e);

http://git-wip-us.apache.org/repos/asf/hive/blob/1785ca00/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java 
b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
index fbf5481..974184f 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
@@ -224,6 +224,11 @@ public class Cleaner extends CompactorThread {
             return null;
           }
         });
+        try {
+          FileSystem.closeAllForUGI(ugi);
+        } catch (IOException exception) {
+          LOG.error("Could not clean up file-system handles for UGI: " + ugi, 
exception + " for " +
+              ci.getFullPartitionName());        }
       }
       txnHandler.markCleaned(ci);
     } catch (Exception e) {

http://git-wip-us.apache.org/repos/asf/hive/blob/1785ca00/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorThread.java
----------------------------------------------------------------------
diff --git 
a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorThread.java 
b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorThread.java
index 3f6b099..859caff 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorThread.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorThread.java
@@ -173,6 +173,11 @@ abstract class CompactorThread extends Thread implements 
MetaStoreThread {
           return null;
         }
       });
+      try {
+        FileSystem.closeAllForUGI(ugi);
+      } catch (IOException exception) {
+        LOG.error("Could not clean up file-system handles for UGI: " + ugi, 
exception);
+      }
 
       if (wrapper.size() == 1) {
         LOG.debug("Running job as " + wrapper.get(0));

http://git-wip-us.apache.org/repos/asf/hive/blob/1785ca00/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java 
b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java
index 2ef06de..9d71c5a 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java
@@ -213,12 +213,19 @@ public class Initiator extends CompactorThread {
       LOG.info("Going to initiate as user " + runAs);
       UserGroupInformation ugi = UserGroupInformation.createProxyUser(runAs,
         UserGroupInformation.getLoginUser());
-      return ugi.doAs(new PrivilegedExceptionAction<CompactionType>() {
+      CompactionType compactionType = ugi.doAs(new 
PrivilegedExceptionAction<CompactionType>() {
         @Override
         public CompactionType run() throws Exception {
           return determineCompactionType(ci, txns, sd);
         }
       });
+      try {
+        FileSystem.closeAllForUGI(ugi);
+      } catch (IOException exception) {
+        LOG.error("Could not clean up file-system handles for UGI: " + ugi, 
exception + " for " +
+            ci.getFullPartitionName());
+      }
+      return compactionType;
     }
   }
 

http://git-wip-us.apache.org/repos/asf/hive/blob/1785ca00/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java 
b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java
index ce03c8e..8dbe3d4 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java
@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.hive.ql.txn.compactor;
 
+import org.apache.hadoop.fs.FileSystem;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.hadoop.hive.common.ValidTxnList;
@@ -171,6 +172,11 @@ public class Worker extends CompactorThread {
                 return null;
               }
             });
+            try {
+              FileSystem.closeAllForUGI(ugi);
+            } catch (IOException exception) {
+              LOG.error("Could not clean up file-system handles for UGI: " + 
ugi, exception + " for " +
+                  ci.getFullPartitionName());            }
           }
           txnHandler.markCompacted(ci);
         } catch (Exception e) {

http://git-wip-us.apache.org/repos/asf/hive/blob/1785ca00/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java
----------------------------------------------------------------------
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java 
b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java
index 7a1a3d2..61df8f0 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java
@@ -49,10 +49,13 @@ import org.junit.Test;
 import org.junit.rules.TestName;
 
 import java.io.File;
+import java.lang.reflect.Field;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Comparator;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -545,6 +548,50 @@ public class TestTxnCommands2 {
     Assert.assertEquals("Unexpected num succeeded", 1, cbs.succeeded);
     Assert.assertEquals("Unexpected num total5", 
hiveConf.getIntVar(HiveConf.ConfVars.COMPACTOR_HISTORY_RETENTION_FAILED) + 1, 
cbs.total);
   }
+
+  /**
+   * Make sure there's no FileSystem$Cache$Key leak due to UGI use
+   * @throws Exception
+   */
+  @Test
+  public void testFileSystemUnCaching() throws Exception {
+    int cacheSizeBefore;
+    int cacheSizeAfter;
+
+    // get the size of cache BEFORE
+    cacheSizeBefore = getFileSystemCacheSize();
+
+    // Insert a row to ACID table
+    runStatementOnDriver("insert into " + Table.ACIDTBL + "(a,b) values(1,2)");
+
+    // Perform a major compaction
+    runStatementOnDriver("alter table " + Table.ACIDTBL + " compact 'major'");
+    runWorker(hiveConf);
+    runCleaner(hiveConf);
+
+    // get the size of cache AFTER
+    cacheSizeAfter = getFileSystemCacheSize();
+
+    Assert.assertEquals(cacheSizeBefore, cacheSizeAfter);
+  }
+
+  private int getFileSystemCacheSize() throws Exception {
+    try {
+      Field cache = FileSystem.class.getDeclaredField("CACHE");
+      cache.setAccessible(true);
+      Object o = cache.get(null); // FileSystem.CACHE
+
+      Field mapField = o.getClass().getDeclaredField("map");
+      mapField.setAccessible(true);
+      Map map = (HashMap)mapField.get(o); // FileSystem.CACHE.map
+
+      return map.size();
+    } catch (NoSuchFieldException e) {
+      System.out.println(e);
+    }
+    return 0;
+  }
+
   private static class CompactionsByState {
     private int attempted;
     private int failed;

Reply via email to