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) {
      cacheableHiveMetaStoreClient = getOrCreate(cacheKey);
      cacheableHiveMetaStoreClient.acquire();
    }
    if (!cacheableHiveMetaStoreClient.isOpen()) {
      synchronized (CACHE_TEARDOWN_LOCK) {
        hiveCache.invalidate(cacheKey);
        cacheableHiveMetaStoreClient.close();
        cacheableHiveMetaStoreClient = getOrCreate(cacheKey);
        cacheableHiveMetaStoreClient.acquire();
      }
    }
    return cacheableHiveMetaStoreClient;
  }
@Test
  public void testCacheHit() throws IOException, MetaException, LoginException {
    HiveClientCache cache = new HiveClientCache(1000);
    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.DYNAMIC_PARTITION_MAX_PARTS, 10);
    HiveClientCache.ICacheableMetaStoreClient client2 = (HiveClientCache.ICacheableMetaStoreClient) cache.get(hiveConf);
    assertNotNull(client2);
    assertSame(client, client2);
    assertEquals(client.getUsers(), client2.getUsers());
    client2.close();
  }