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;
  }
@Ignore("hangs indefinitely")
  @Test
  public void testHMSCBreakability() throws IOException, MetaException, LoginException, TException, AlreadyExistsException,
      InvalidObjectException, NoSuchObjectException, InterruptedException {
    // Setup
    LocalMetaServer metaServer = new LocalMetaServer();
    metaServer.start();

    final HiveClientCache cache = new HiveClientCache(1000);
    HiveClientCache.CacheableHiveMetaStoreClient client =
        (HiveClientCache.CacheableHiveMetaStoreClient) cache.get(metaServer.getHiveConf());

    assertTrue(client.isOpen());

    final String DB_NAME = "test_db";
    final String LONG_TABLE_NAME = "long_table_name_" + new BigInteger(200, new Random()).toString(2);

    try {
      client.dropTable(DB_NAME, LONG_TABLE_NAME);
    } catch (Exception e) {
    }
    try {
      client.dropDatabase(DB_NAME);
    } catch (Exception e) {
    }

    client.createDatabase(new Database(DB_NAME, "", null, null));

    List<FieldSchema> fields = new ArrayList<FieldSchema>();
    fields.add(new FieldSchema("colname", serdeConstants.STRING_TYPE_NAME, ""));
    Table tbl = new Table();
    tbl.setDbName(DB_NAME);
    tbl.setTableName(LONG_TABLE_NAME);
    StorageDescriptor sd = new StorageDescriptor();
    sd.setCols(fields);
    tbl.setSd(sd);
    sd.setSerdeInfo(new SerDeInfo());

    // Break the client
    try {
      client.createTable(tbl);
      fail("Exception was expected while creating table with long name");
    } catch (Exception e) {
    }

    assertFalse(client.isOpen());
    metaServer.shutDown();
  }