hive-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mit...@apache.org
Subject hive git commit: HIVE-13754: Fix resource leak in HiveClientCache (Chris Drome, via Mithun Radhakrishnan)
Date Wed, 10 Aug 2016 17:12:07 GMT
Repository: hive
Updated Branches:
  refs/heads/master 41cb67e61 -> 925f19552


HIVE-13754: Fix resource leak in HiveClientCache (Chris Drome, via Mithun Radhakrishnan)


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

Branch: refs/heads/master
Commit: 925f195523aba1e2bfbfebf06c6b87fafa4dbd67
Parents: 41cb67e
Author: Mithun RK <mithunr@yahoo-inc.com>
Authored: Wed Aug 10 10:10:32 2016 -0700
Committer: Mithun RK <mithunr@yahoo-inc.com>
Committed: Wed Aug 10 10:11:38 2016 -0700

----------------------------------------------------------------------
 .../hive/hcatalog/common/HCatConstants.java     |  10 +
 .../apache/hive/hcatalog/common/HCatUtil.java   |  13 +-
 .../hive/hcatalog/common/HiveClientCache.java   | 219 ++++++++++++++-----
 .../hcatalog/common/TestHiveClientCache.java    |  14 +-
 .../hadoop/hive/metastore/IMetaStoreClient.java |   2 +
 .../hive/metastore/RetryingMetaStoreClient.java |  18 +-
 .../hive/metastore/annotation/NoReconnect.java  |  29 +++
 7 files changed, 233 insertions(+), 72 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hive/blob/925f1955/hcatalog/core/src/main/java/org/apache/hive/hcatalog/common/HCatConstants.java
----------------------------------------------------------------------
diff --git a/hcatalog/core/src/main/java/org/apache/hive/hcatalog/common/HCatConstants.java
b/hcatalog/core/src/main/java/org/apache/hive/hcatalog/common/HCatConstants.java
index d165e7e..72930eb 100644
--- a/hcatalog/core/src/main/java/org/apache/hive/hcatalog/common/HCatConstants.java
+++ b/hcatalog/core/src/main/java/org/apache/hive/hcatalog/common/HCatConstants.java
@@ -61,11 +61,21 @@ public final class HCatConstants {
 
   // hcatalog specific configurations, that can be put in hive-site.xml
   public static final String HCAT_HIVE_CLIENT_EXPIRY_TIME = "hcatalog.hive.client.cache.expiry.time";
+
   // config parameter that suggests to hcat that metastore clients not be cached - default
is false
   // this parameter allows highly-parallel hcat usescases to not gobble up too many connections
that
   // sit in the cache, while not in use.
   public static final String HCAT_HIVE_CLIENT_DISABLE_CACHE = "hcatalog.hive.client.cache.disabled";
 
+  // Indicates the initial capacity of the cache.
+  public static final String HCAT_HIVE_CLIENT_CACHE_INITIAL_CAPACITY = "hcatalog.hive.client.cache.initial.capacity";
+
+  // Indicates the maximum capacity of the cache. Minimum value should be the number of threads.
+  public static final String HCAT_HIVE_CLIENT_CACHE_MAX_CAPACITY = "hcatalog.hive.client.cache.max.capacity";
+
+  // Indicates whether cache statistics should be collected.
+  public static final String HCAT_HIVE_CLIENT_CACHE_STATS_ENABLED = "hcatalog.hive.client.cache.stats.enabled";
+
   private HCatConstants() { // restrict instantiation
   }
 

http://git-wip-us.apache.org/repos/asf/hive/blob/925f1955/hcatalog/core/src/main/java/org/apache/hive/hcatalog/common/HCatUtil.java
----------------------------------------------------------------------
diff --git a/hcatalog/core/src/main/java/org/apache/hive/hcatalog/common/HCatUtil.java b/hcatalog/core/src/main/java/org/apache/hive/hcatalog/common/HCatUtil.java
index f3bfcfa..8b927af 100644
--- a/hcatalog/core/src/main/java/org/apache/hive/hcatalog/common/HCatUtil.java
+++ b/hcatalog/core/src/main/java/org/apache/hive/hcatalog/common/HCatUtil.java
@@ -573,15 +573,10 @@ public class HCatUtil {
    */
   @Deprecated
   public static HiveMetaStoreClient getHiveClient(HiveConf hiveConf) throws MetaException,
IOException {
-    IMetaStoreClient imsc = getHiveMetastoreClient(hiveConf);
-    // Try piggybacking on the function that returns IMSC. Current implementation of the
IMSC cache
-    // has CacheableMetaStoreClients, which are HMSC, so we can return them as-is. If not,
it's okay
-    // for us to ignore the caching aspect and return a vanilla HMSC.
-    if (imsc instanceof HiveMetaStoreClient){
-      return (HiveMetaStoreClient)imsc;
-    } else {
-      return new HiveMetaStoreClient(hiveConf);
-    }
+    LOG.warn("HCatUtil.getHiveClient is unsafe and can be a resource leak depending on HMSC
"
+        + "implementation and caching mechanism. Use HCatUtil.getHiveMetastoreClient instead.");
+
+    return new HiveMetaStoreClient(hiveConf);
   }
 
   public static void closeHiveClientQuietly(IMetaStoreClient client) {

http://git-wip-us.apache.org/repos/asf/hive/blob/925f1955/hcatalog/core/src/main/java/org/apache/hive/hcatalog/common/HiveClientCache.java
----------------------------------------------------------------------
diff --git a/hcatalog/core/src/main/java/org/apache/hive/hcatalog/common/HiveClientCache.java
b/hcatalog/core/src/main/java/org/apache/hive/hcatalog/common/HiveClientCache.java
index 9b30703..51f516e 100644
--- a/hcatalog/core/src/main/java/org/apache/hive/hcatalog/common/HiveClientCache.java
+++ b/hcatalog/core/src/main/java/org/apache/hive/hcatalog/common/HiveClientCache.java
@@ -32,11 +32,13 @@ import javax.security.auth.login.LoginException;
 
 import org.apache.commons.lang.builder.EqualsBuilder;
 import org.apache.commons.lang.builder.HashCodeBuilder;
+import org.apache.hadoop.hive.common.classification.InterfaceAudience;
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
 import org.apache.hadoop.hive.metastore.IMetaStoreClient;
 import org.apache.hadoop.hive.metastore.RetryingMetaStoreClient;
 import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.annotation.NoReconnect;
 import org.apache.hadoop.hive.shims.ShimLoader;
 import org.apache.hadoop.hive.shims.Utils;
 import org.apache.hadoop.security.UserGroupInformation;
@@ -55,8 +57,11 @@ import com.google.common.util.concurrent.ThreadFactoryBuilder;
  */
 class HiveClientCache {
   public final static int DEFAULT_HIVE_CACHE_EXPIRY_TIME_SECONDS = 2 * 60;
+  public final static int DEFAULT_HIVE_CACHE_INITIAL_CAPACITY = 50;
+  public final static int DEFAULT_HIVE_CACHE_MAX_CAPACITY = 50;
+  public final static boolean DEFAULT_HIVE_CLIENT_CACHE_STATS_ENABLED = false;
 
-  final private Cache<HiveClientCacheKey, ICacheableMetaStoreClient> hiveCache;
+  private final Cache<HiveClientCacheKey, ICacheableMetaStoreClient> hiveCache;
   private static final Logger LOG = LoggerFactory.getLogger(HiveClientCache.class);
   private final int timeout;
   // This lock is used to make sure removalListener won't close a client that is being contemplated
for returning by get()
@@ -66,6 +71,8 @@ class HiveClientCache {
 
   private final ScheduledFuture<?> cleanupHandle; // used to cleanup cache
 
+  private boolean enableStats;
+
   // Since HiveMetaStoreClient is not threadsafe, hive clients are not  shared across threads.
   // Thread local variable containing each thread's unique ID, is used as one of the keys
for the cache
   // causing each thread to get a different client even if the hiveConf is same.
@@ -86,20 +93,78 @@ class HiveClientCache {
   }
 
   public HiveClientCache(HiveConf hiveConf) {
-    this(hiveConf.getInt(HCatConstants.HCAT_HIVE_CLIENT_EXPIRY_TIME,
-        DEFAULT_HIVE_CACHE_EXPIRY_TIME_SECONDS));
+    this(hiveConf.getInt(HCatConstants.HCAT_HIVE_CLIENT_EXPIRY_TIME, DEFAULT_HIVE_CACHE_EXPIRY_TIME_SECONDS),
+        hiveConf.getInt(HCatConstants.HCAT_HIVE_CLIENT_CACHE_INITIAL_CAPACITY, DEFAULT_HIVE_CACHE_INITIAL_CAPACITY),
+        hiveConf.getInt(HCatConstants.HCAT_HIVE_CLIENT_CACHE_MAX_CAPACITY, DEFAULT_HIVE_CACHE_MAX_CAPACITY),
+        hiveConf.getBoolean(HCatConstants.HCAT_HIVE_CLIENT_CACHE_STATS_ENABLED, DEFAULT_HIVE_CLIENT_CACHE_STATS_ENABLED));
+
   }
+
   /**
-   * @param timeout the length of time in seconds after a client is created that it should
be automatically removed
+   * @deprecated This constructor will be made private or removed as more configuration properties
are required.
    */
+  @Deprecated
   public HiveClientCache(final int timeout) {
+    this(timeout, DEFAULT_HIVE_CACHE_INITIAL_CAPACITY, DEFAULT_HIVE_CACHE_MAX_CAPACITY, DEFAULT_HIVE_CLIENT_CACHE_STATS_ENABLED);
+  }
+
+  /**
+   * @param timeout the length of time in seconds after a client is created that it should
be automatically removed
+   */
+  private HiveClientCache(final int timeout, final int initialCapacity, final int maxCapacity,
final boolean enableStats) {
     this.timeout = timeout;
-    RemovalListener<HiveClientCacheKey, ICacheableMetaStoreClient> removalListener
=
+    this.enableStats = enableStats;
+
+    LOG.info("Initializing cache: eviction-timeout=" + timeout + " initial-capacity=" + initialCapacity
+ " maximum-capacity=" + maxCapacity);
+
+    CacheBuilder builder = CacheBuilder.newBuilder()
+      .initialCapacity(initialCapacity)
+      .maximumSize(maxCapacity)
+      .expireAfterAccess(timeout, TimeUnit.SECONDS)
+      .removalListener(createRemovalListener());
+
+    /*
+     * Guava versions <12.0 have stats collection enabled by default and do not expose
a recordStats method.
+     * Check for newer versions of the library and ensure that stats collection is enabled
by default.
+     */
+    try {
+      java.lang.reflect.Method m = builder.getClass().getMethod("recordStats", null);
+      m.invoke(builder, null);
+    } catch (NoSuchMethodException e) {
+      LOG.debug("Using a version of guava <12.0. Stats collection is enabled by default.");
+    } catch (Exception e) {
+      LOG.warn("Unable to invoke recordStats method.", e);
+    }
+
+    this.hiveCache = builder.build();
+
+    /*
+     * We need to use a cleanup interval, which is how often the cleanup thread will kick
in
+     * and go do a check to see if any of the connections can be expired. We don't want to
+     * do this too often, because it'd be like having a mini-GC going off every so often,
+     * so we limit it to a minimum of DEFAULT_HIVE_CACHE_EXPIRY_TIME_SECONDS. If the client
+     * has explicitly set a larger timeout on the cache, though, we respect that, and use
that
+     */
+    long cleanupInterval = timeout > DEFAULT_HIVE_CACHE_EXPIRY_TIME_SECONDS ? timeout
: DEFAULT_HIVE_CACHE_EXPIRY_TIME_SECONDS;
+
+    this.cleanupHandle = createCleanupThread(cleanupInterval);
+
+    createShutdownHook();
+  }
+
+  private RemovalListener<HiveClientCacheKey, ICacheableMetaStoreClient> createRemovalListener()
{
+    RemovalListener<HiveClientCacheKey, ICacheableMetaStoreClient> listener =
       new RemovalListener<HiveClientCacheKey, ICacheableMetaStoreClient>() {
         @Override
         public void onRemoval(RemovalNotification<HiveClientCacheKey, ICacheableMetaStoreClient>
notification) {
           ICacheableMetaStoreClient hiveMetaStoreClient = notification.getValue();
           if (hiveMetaStoreClient != null) {
+            if (LOG.isDebugEnabled()) {
+              LOG.debug("Evicting client: " + Integer.toHexString(System.identityHashCode(hiveMetaStoreClient)));
+            }
+
+            // TODO: This global lock may not be necessary as all concurrent methods in ICacheableMetaStoreClient
+            // are synchronized.
             synchronized (CACHE_TEARDOWN_LOCK) {
               hiveMetaStoreClient.setExpiredFromCache();
               hiveMetaStoreClient.tearDownIfUnused();
@@ -107,33 +172,20 @@ class HiveClientCache {
           }
         }
       };
-    hiveCache = CacheBuilder.newBuilder()
-      .expireAfterWrite(timeout, TimeUnit.SECONDS)
-      .removalListener(removalListener)
-      .build();
 
+    return listener;
+  }
+
+  private ScheduledFuture<?> createCleanupThread(long interval) {
     // Add a maintenance thread that will attempt to trigger a cache clean continuously
     Runnable cleanupThread = new Runnable() {
       @Override
       public void run() {
-        hiveCache.cleanUp();
+        cleanup();
       }
     };
 
     /**
-     * We need to use a cleanup interval, which is how often the cleanup thread will kick
in
-     * and go do a check to see if any of the connections can be expired. We don't want to
-     * do this too often, because it'd be like having a mini-GC going off every so often,
-     * so we limit it to a minimum of DEFAULT_HIVE_CACHE_EXPIRY_TIME_SECONDS. If the client
-     * has explicitly set a larger timeout on the cache, though, we respect that, and use
that
-     */
-    long cleanupInterval = DEFAULT_HIVE_CACHE_EXPIRY_TIME_SECONDS;
-
-    if (timeout > cleanupInterval){
-      cleanupInterval = timeout;
-    }
-
-    /**
      * Create the cleanup handle. In addition to cleaning up every cleanupInterval, we add
      * a slight offset, so that the very first time it runs, it runs with a slight delay,
so
      * as to catch any other connections that were closed when the first timeout happened.
@@ -142,13 +194,14 @@ class HiveClientCache {
      * it can be cleaned every max(DEFAULT_HIVE_CACHE_EXPIRY_TIME_SECONDS,timeout) seconds
      */
     ThreadFactory daemonThreadFactory = (new ThreadFactoryBuilder()).setDaemon(true)
-        .setNameFormat("HiveClientCache-cleaner-%d").build();
-
-    cleanupHandle = Executors.newScheduledThreadPool(1, daemonThreadFactory).scheduleWithFixedDelay(
-        cleanupThread,
-        timeout + 5, cleanupInterval, TimeUnit.SECONDS);
+      .setNameFormat("HiveClientCache-cleaner-%d")
+      .build();
 
+    return Executors.newScheduledThreadPool(1, daemonThreadFactory)
+      .scheduleWithFixedDelay(cleanupThread, timeout + 5, interval, TimeUnit.SECONDS);
+  }
 
+  private void createShutdownHook() {
     // Add a shutdown hook for cleanup, if there are elements remaining in the cache which
were not cleaned up.
     // This is the best effort approach. Ignore any error while doing so. Notice that most
of the clients
     // would get cleaned up via either the removalListener or the close() call, only the
active clients
@@ -163,6 +216,7 @@ class HiveClientCache {
         closeAllClientsQuietly();
       }
     };
+
     Runtime.getRuntime().addShutdownHook(cleanupHiveClientShutdownThread);
   }
 
@@ -178,10 +232,19 @@ class HiveClientCache {
     } catch (Exception e) {
       LOG.warn("Clean up of hive clients in the cache failed. Ignored", e);
     }
+
+    if (this.enableStats) {
+      LOG.info("Cache statistics after shutdown: size=" + hiveCache.size() + " " + hiveCache.stats());
+    }
   }
 
   public void cleanup() {
+    // TODO: periodically reload a new HiveConf to check if stats reporting is enabled.
     hiveCache.cleanUp();
+
+    if (enableStats) {
+      LOG.info("Cache statistics after cleanup: size=" + hiveCache.size() + " " + hiveCache.stats());
+    }
   }
 
   /**
@@ -193,9 +256,10 @@ class HiveClientCache {
    * @throws IOException
    * @throws LoginException
    */
-  public ICacheableMetaStoreClient get(final HiveConf hiveConf) throws MetaException, IOException,
LoginException {
+  public IMetaStoreClient get(final HiveConf hiveConf) throws MetaException, IOException,
LoginException {
     final HiveClientCacheKey cacheKey = HiveClientCacheKey.fromHiveConf(hiveConf, getThreadId());
     ICacheableMetaStoreClient cacheableHiveMetaStoreClient = null;
+
     // the hmsc is not shared across threads. So the only way it could get closed while we
are doing healthcheck
     // is if removalListener closes it. The synchronization takes care that removalListener
won't do it
     synchronized (CACHE_TEARDOWN_LOCK) {
@@ -255,7 +319,7 @@ class HiveClientCache {
    * UserGroupInformation and metaStoreURIs are same. This function can evolve to express
    * the cases when HiveConf is different but the same hiveMetaStoreClient can be used
    */
-  public static class HiveClientCacheKey {
+  static class HiveClientCacheKey {
     final private String metaStoreURIs;
     final private UserGroupInformation ugi;
     final private HiveConf hiveConf;
@@ -295,24 +359,38 @@ class HiveClientCache {
         append(ugi).
         append(threadId).toHashCode();
     }
+
+    @Override
+    public String toString() {
+      return "HiveClientCacheKey: uri=" + this.metaStoreURIs + " ugi=" + this.ugi + " thread="
+ this.threadId; 
+    }
   }
 
+  @InterfaceAudience.Private
   public interface ICacheableMetaStoreClient extends IMetaStoreClient {
-
+    @NoReconnect
     void acquire();
 
-    void release();
-
+    @NoReconnect
     void setExpiredFromCache();
 
+    @NoReconnect
     AtomicInteger getUsers();
 
+    @NoReconnect
     boolean isClosed();
 
+    /**
+     * @deprecated This method is not used internally and should not be visible through HCatClient.create.
+     */
+    @Deprecated
+    @NoReconnect
     boolean isOpen();
 
+    @NoReconnect
     void tearDownIfUnused();
 
+    @NoReconnect
     void tearDown();
   }
 
@@ -324,26 +402,48 @@ class HiveClientCache {
     private final AtomicInteger users = new AtomicInteger(0);
     private volatile boolean expiredFromCache = false;
     private boolean isClosed = false;
-    private final long expiryTime;
-    private static final int EXPIRY_TIME_EXTENSION_IN_MILLIS = 60 * 1000;
 
     CacheableHiveMetaStoreClient(final HiveConf conf, final Integer timeout, Boolean allowEmbedded)
         throws MetaException {
       super(conf, null, allowEmbedded);
-      // Extend the expiry time with some extra time on top of guava expiry time to make
sure
-      // that items closed() are for sure expired and would never be returned by guava.
-      this.expiryTime = System.currentTimeMillis() + timeout * 1000 + EXPIRY_TIME_EXTENSION_IN_MILLIS;
     }
 
-    public void acquire() {
+    /**
+     * Increments the user count and optionally renews the expiration time.
+     * <code>renew</code> should correspond with the expiration policy of the
cache.
+     * When the policy is <code>expireAfterAccess</code>, the expiration time
should be extended.
+     * When the policy is <code>expireAfterWrite</code>, the expiration time
should not be extended.
+     * A mismatch with the policy will lead to closing the connection unnecessarily after
the initial
+     * expiration time is generated.
+     * @param renew whether the expiration time should be extended.
+     */
+    public synchronized void acquire() {
       users.incrementAndGet();
+      if (users.get() > 1) {
+        LOG.warn("Unexpected increment of user count beyond one: " + users.get() + " " +
this);
+      }
     }
 
-    public void release() {
-      users.decrementAndGet();
+    /**
+     * Decrements the user count.
+     */
+    private void release() {
+      if (users.get() > 0) {
+        users.decrementAndGet();
+      } else {
+        LOG.warn("Unexpected attempt to decrement user count of zero: " + users.get() + "
" + this);
+      }
     }
 
-    public void setExpiredFromCache() {
+    /**
+     * Communicate to the client that it is no longer in the cache.
+     * The expiration time should be voided to allow the connection to be closed at the first
opportunity.
+     */
+    public synchronized void setExpiredFromCache() {
+      if (users.get() != 0) {
+        LOG.warn("Evicted client has non-zero user count: " + users.get());
+      }
+
       expiredFromCache = true;
     }
 
@@ -363,6 +463,7 @@ class HiveClientCache {
      * invalid data renders the client unusable for future use (example: create a table with
very long table name)
      * @return
      */
+    @Deprecated
     public boolean isOpen() {
       try {
         // Look for an unlikely database name and see if either MetaException or TException
is thrown
@@ -378,28 +479,31 @@ class HiveClientCache {
      * This *MUST* be called by anyone who uses this client.
      */
     @Override
-    public void close() {
+    public synchronized void close() {
       release();
-      if (System.currentTimeMillis() >= expiryTime)
-        setExpiredFromCache();
       tearDownIfUnused();
     }
 
     /**
-     * Tear down only if
-     *  1. There are no active user
-     *  2. It has expired from the cache
+     * Attempt to tear down the client connection.
+     * The connection will be closed if the following conditions hold:
+     *  1. There are no active user holding the client.
+     *  2. The client has been evicted from the cache.
      */
-    public void tearDownIfUnused() {
+    public synchronized void tearDownIfUnused() {
+      if (users.get() != 0) {
+        LOG.warn("Non-zero user count preventing client tear down: users=" + users.get()
+ " expired=" + expiredFromCache);
+      }
+
       if (users.get() == 0 && expiredFromCache) {
         this.tearDown();
       }
     }
 
     /**
-     * Close if not closed already
+     * Close the underlying objects irrespective of whether they are in use or not.
      */
-    public synchronized void tearDown() {
+    public void tearDown() {
       try {
         if (!isClosed) {
           super.close();
@@ -410,12 +514,23 @@ class HiveClientCache {
       }
     }
 
+    @Override
+    public String toString() {
+      return "HCatClient: thread: " + Thread.currentThread().getId() + " users=" + users.get()
+        + " expired=" + expiredFromCache + " closed=" + isClosed;
+    }
+
     /**
-     * Last effort to clean up, may not even get called.
+     * GC is attempting to destroy the object.
+     * No one references this client anymore, so it can be torn down without worrying about
user counts.
      * @throws Throwable
      */
     @Override
     protected void finalize() throws Throwable {
+      if (users.get() != 0) {
+        LOG.warn("Closing client with non-zero user count: users=" + users.get() + " expired="
+ expiredFromCache);
+      }
+
       try {
         this.tearDown();
       } finally {

http://git-wip-us.apache.org/repos/asf/hive/blob/925f1955/hcatalog/core/src/test/java/org/apache/hive/hcatalog/common/TestHiveClientCache.java
----------------------------------------------------------------------
diff --git a/hcatalog/core/src/test/java/org/apache/hive/hcatalog/common/TestHiveClientCache.java
b/hcatalog/core/src/test/java/org/apache/hive/hcatalog/common/TestHiveClientCache.java
index b2c9c7a..c77bc48 100644
--- a/hcatalog/core/src/test/java/org/apache/hive/hcatalog/common/TestHiveClientCache.java
+++ b/hcatalog/core/src/test/java/org/apache/hive/hcatalog/common/TestHiveClientCache.java
@@ -41,6 +41,7 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
@@ -77,14 +78,15 @@ public class TestHiveClientCache {
   @Test
   public void testCacheHit() throws IOException, MetaException, LoginException {
     HiveClientCache cache = new HiveClientCache(1000);
-    HiveClientCache.ICacheableMetaStoreClient client = cache.get(hiveConf);
+    HiveClientCache.ICacheableMetaStoreClient client = (HiveClientCache.ICacheableMetaStoreClient)
cache.get(hiveConf);
     assertNotNull(client);
     client.close(); // close shouldn't matter
 
     // Setting a non important configuration should return the same client only
     hiveConf.setIntVar(HiveConf.ConfVars.DYNAMICPARTITIONMAXPARTS, 10);
-    HiveClientCache.ICacheableMetaStoreClient client2 = cache.get(hiveConf);
+    HiveClientCache.ICacheableMetaStoreClient client2 = (HiveClientCache.ICacheableMetaStoreClient)
cache.get(hiveConf);
     assertNotNull(client2);
+    assertSame(client, client2);
     assertEquals(client.getUsers(), client2.getUsers());
     client2.close();
   }
@@ -109,11 +111,11 @@ public class TestHiveClientCache {
   @Test
   public void testCacheExpiry() throws IOException, MetaException, LoginException, InterruptedException
{
     HiveClientCache cache = new HiveClientCache(1);
-    HiveClientCache.ICacheableMetaStoreClient client = cache.get(hiveConf);
+    HiveClientCache.ICacheableMetaStoreClient client = (HiveClientCache.ICacheableMetaStoreClient)
cache.get(hiveConf);
     assertNotNull(client);
 
     Thread.sleep(2500);
-    HiveClientCache.ICacheableMetaStoreClient client2 = cache.get(hiveConf);
+    HiveClientCache.ICacheableMetaStoreClient client2 = (HiveClientCache.ICacheableMetaStoreClient)
cache.get(hiveConf);
     client.close();
     assertTrue(client.isClosed()); // close() after *expiry time* and *a cache access* should
 have tore down the client
 
@@ -154,9 +156,9 @@ public class TestHiveClientCache {
   @Test
   public void testCloseAllClients() throws IOException, MetaException, LoginException {
     final HiveClientCache cache = new HiveClientCache(1000);
-    HiveClientCache.ICacheableMetaStoreClient client1 = cache.get(hiveConf);
+    HiveClientCache.ICacheableMetaStoreClient client1 = (HiveClientCache.ICacheableMetaStoreClient)
cache.get(hiveConf);
     hiveConf.setVar(HiveConf.ConfVars.METASTOREURIS, " "); // URIs are checked for string
equivalence, even spaces make them different
-    HiveClientCache.ICacheableMetaStoreClient client2 = cache.get(hiveConf);
+    HiveClientCache.ICacheableMetaStoreClient client2 = (HiveClientCache.ICacheableMetaStoreClient)
cache.get(hiveConf);
     cache.closeAllClientsQuietly();
     assertTrue(client1.isClosed());
     assertTrue(client2.isClosed());

http://git-wip-us.apache.org/repos/asf/hive/blob/925f1955/metastore/src/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java
----------------------------------------------------------------------
diff --git a/metastore/src/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java b/metastore/src/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java
index b6fe502..8dc4e28 100644
--- a/metastore/src/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java
+++ b/metastore/src/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java
@@ -25,6 +25,7 @@ import org.apache.hadoop.hive.common.classification.InterfaceAudience;
 import org.apache.hadoop.hive.common.classification.InterfaceAudience.Public;
 import org.apache.hadoop.hive.common.classification.InterfaceStability.Evolving;
 import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.annotation.NoReconnect;
 import org.apache.hadoop.hive.metastore.api.AggrStats;
 import org.apache.hadoop.hive.metastore.api.AlreadyExistsException;
 import org.apache.hadoop.hive.metastore.api.ColumnStatistics;
@@ -127,6 +128,7 @@ public interface IMetaStoreClient {
   /**
    * close connection to meta store
    */
+  @NoReconnect
   void close();
 
   /**

http://git-wip-us.apache.org/repos/asf/hive/blob/925f1955/metastore/src/java/org/apache/hadoop/hive/metastore/RetryingMetaStoreClient.java
----------------------------------------------------------------------
diff --git a/metastore/src/java/org/apache/hadoop/hive/metastore/RetryingMetaStoreClient.java
b/metastore/src/java/org/apache/hadoop/hive/metastore/RetryingMetaStoreClient.java
index c0d9c0c..4895bff 100644
--- a/metastore/src/java/org/apache/hadoop/hive/metastore/RetryingMetaStoreClient.java
+++ b/metastore/src/java/org/apache/hadoop/hive/metastore/RetryingMetaStoreClient.java
@@ -32,6 +32,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.hadoop.hive.common.classification.InterfaceAudience.Public;
 import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.annotation.NoReconnect;
 import org.apache.hadoop.hive.metastore.api.MetaException;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.thrift.TApplicationException;
@@ -139,13 +140,20 @@ public class RetryingMetaStoreClient implements InvocationHandler {
     Object ret = null;
     int retriesMade = 0;
     TException caughtException = null;
+
+    boolean allowReconnect = ! method.isAnnotationPresent(NoReconnect.class);
+
     while (true) {
       try {
         reloginExpiringKeytabUser();
-        if (retriesMade > 0 || hasConnectionLifeTimeReached(method)) {
-          base.reconnect();
-          lastConnectionTime = System.currentTimeMillis();
+
+        if (allowReconnect) {
+          if (retriesMade > 0 || hasConnectionLifeTimeReached(method)) {
+            base.reconnect();
+            lastConnectionTime = System.currentTimeMillis();
+          }
         }
+
         if (metaCallTimeMap == null) {
           ret = method.invoke(base, args);
         } else {
@@ -228,10 +236,10 @@ public class RetryingMetaStoreClient implements InvocationHandler {
   }
 
   private boolean hasConnectionLifeTimeReached(Method method) {
-    if (connectionLifeTimeInMillis <= 0 || localMetaStore ||
-        method.getName().equalsIgnoreCase("close")) {
+    if (connectionLifeTimeInMillis <= 0 || localMetaStore) {
       return false;
     }
+
     boolean shouldReconnect =
         (System.currentTimeMillis() - lastConnectionTime) >= connectionLifeTimeInMillis;
     if (LOG.isDebugEnabled()) {

http://git-wip-us.apache.org/repos/asf/hive/blob/925f1955/metastore/src/java/org/apache/hadoop/hive/metastore/annotation/NoReconnect.java
----------------------------------------------------------------------
diff --git a/metastore/src/java/org/apache/hadoop/hive/metastore/annotation/NoReconnect.java
b/metastore/src/java/org/apache/hadoop/hive/metastore/annotation/NoReconnect.java
new file mode 100644
index 0000000..edf0831
--- /dev/null
+++ b/metastore/src/java/org/apache/hadoop/hive/metastore/annotation/NoReconnect.java
@@ -0,0 +1,29 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.hadoop.hive.metastore.annotation;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Target({ElementType.METHOD})
+@Retention(RetentionPolicy.RUNTIME)
+public @interface NoReconnect {
+}


Mime
View raw message