@Test public void testSimple() throws Exception { HiveTable tab21 = new HiveTable("tab21"); HiveTable tab31 = new HiveTable("tab31").add("part311").add("part312"); HiveDb db3 = new HiveDb("db3", Lists.newArrayList(tab31)); HiveDb db2 = new HiveDb("db2", Lists.newArrayList(tab21)); HiveDb db1 = new HiveDb("db1"); HiveSnapshot snap = new HiveSnapshot().add(db1).add(db2).add(db3); MockClient c = new MockClient(snap); Map<String, Collection<String>> update; try(FullUpdateInitializer cacheInitializer = new FullUpdateInitializer(new MockHMSClientFactory(c), conf)) { update = cacheInitializer.getFullHMSSnapshot(); } Assert.assertEquals(5, update.size()); Assert.assertEquals(Sets.newHashSet("db1"), update.get("db1")); Assert.assertEquals(Sets.newHashSet("db2"), update.get("db2")); Assert.assertEquals(Sets.newHashSet("db3"), update.get("db3")); Assert.assertEquals(Sets.newHashSet("db2/tab21"), update.get("db2.tab21")); Assert.assertEquals(Sets.newHashSet("db3/tab31", "db3/tab31/part311", "db3/tab31/part312"), update.get("db3.tab31")); }
@SuppressWarnings("squid:S00112") Map<String, Collection<String>> getFullHMSSnapshot() throws Exception { List<String> allDbStr; HMSClient c = null; try (HMSClient client = clientFactory.connect()) { c = client; LOGGER.debug("Fetch all db names"); allDbStr = client.getClient().getAllDatabases(); SentryMetrics.getInstance().databaseCount.inc(allDbStr.size()); LOGGER.info("Total number of db names fetched = {}", allDbStr.size()); } catch (Exception e) { if (c != null) { c.invalidate(); } throw e; } for (String dbName : allDbStr) { results.add(threadPool.submit(new DbTask(dbName))); } Map<String, Collection<String>> fullSnapshot = new HashMap<>(); long printMessageTime = System.currentTimeMillis(); while (!results.isEmpty()) { Future<CallResult> result = results.pop(); CallResult callResult = result.get(); if (!callResult.success()) { throw callResult.getFailure(); } Map<String, Set<String>> objectMapping = callResult.getObjectMapping().getObjects(); for (Map.Entry<String, Set<String>> entry: objectMapping.entrySet()) { String key = entry.getKey(); Set<String> val = entry.getValue(); Set<String> existingSet = (Set<String>)fullSnapshot.get(key); if (existingSet == null) { fullSnapshot.put(key, val); continue; } existingSet.addAll(val); } if(System.currentTimeMillis() - printMessageTime > printSnapshotFetchTimeInterval) { long totalNumberOfDatabases = SentryMetrics.getInstance().databaseCount.getCount(); long totalNumberOfTables = SentryMetrics.getInstance().tableCount.getCount(); long totalNumberOfPartitions = SentryMetrics.getInstance().partitionCount.getCount(); double percentageDatabasesFetched = totalNumberOfDatabases > 0? ((double)totalNumberOfDatabasesFetched/totalNumberOfDatabases)*100:0; double percentageTablesFetched = totalNumberOfTables > 0? ((double)totalNumberOfTablesFetched/totalNumberOfTables)*100:0; double percentagePartitionsFetched = totalNumberOfPartitions > 0? ((double)totalNumberOfPartitionsFetched/totalNumberOfPartitions)*100:0; String snapshotFetchStatusString = String.format("Fetching full hms snapshot: databases fetched=%d (%.2f%%); " + "tables fetched=%d (%.2f%%); partitions fetched=%d (%.2f%%); total number of databases=%d; " + "total number of tables=%d total number of partitions=%d", totalNumberOfDatabasesFetched, percentageDatabasesFetched, totalNumberOfTablesFetched, percentageTablesFetched, totalNumberOfPartitionsFetched, percentagePartitionsFetched, totalNumberOfDatabases, totalNumberOfTables, totalNumberOfPartitions); LOGGER.info(snapshotFetchStatusString); printMessageTime = System.currentTimeMillis(); } } return fullSnapshot; }
FullUpdateInitializer implements AutoCloseable { @SuppressWarnings("squid:S00112") Map<String, Collection<String>> getFullHMSSnapshot() throws Exception { List<String> allDbStr; HMSClient c = null; try (HMSClient client = clientFactory.connect()) { c = client; LOGGER.debug("Fetch all db names"); allDbStr = client.getClient().getAllDatabases(); SentryMetrics.getInstance().databaseCount.inc(allDbStr.size()); LOGGER.info("Total number of db names fetched = {}", allDbStr.size()); } catch (Exception e) { if (c != null) { c.invalidate(); } throw e; } for (String dbName : allDbStr) { results.add(threadPool.submit(new DbTask(dbName))); } Map<String, Collection<String>> fullSnapshot = new HashMap<>(); long printMessageTime = System.currentTimeMillis(); while (!results.isEmpty()) { Future<CallResult> result = results.pop(); CallResult callResult = result.get(); if (!callResult.success()) { throw callResult.getFailure(); } Map<String, Set<String>> objectMapping = callResult.getObjectMapping().getObjects(); for (Map.Entry<String, Set<String>> entry: objectMapping.entrySet()) { String key = entry.getKey(); Set<String> val = entry.getValue(); Set<String> existingSet = (Set<String>)fullSnapshot.get(key); if (existingSet == null) { fullSnapshot.put(key, val); continue; } existingSet.addAll(val); } if(System.currentTimeMillis() - printMessageTime > printSnapshotFetchTimeInterval) { long totalNumberOfDatabases = SentryMetrics.getInstance().databaseCount.getCount(); long totalNumberOfTables = SentryMetrics.getInstance().tableCount.getCount(); long totalNumberOfPartitions = SentryMetrics.getInstance().partitionCount.getCount(); double percentageDatabasesFetched = totalNumberOfDatabases > 0? ((double)totalNumberOfDatabasesFetched/totalNumberOfDatabases)*100:0; double percentageTablesFetched = totalNumberOfTables > 0? ((double)totalNumberOfTablesFetched/totalNumberOfTables)*100:0; double percentagePartitionsFetched = totalNumberOfPartitions > 0? ((double)totalNumberOfPartitionsFetched/totalNumberOfPartitions)*100:0; String snapshotFetchStatusString = String.format("Fetching full hms snapshot: databases fetched=%d (%.2f%%); " + "tables fetched=%d (%.2f%%); partitions fetched=%d (%.2f%%); total number of databases=%d; " + "total number of tables=%d total number of partitions=%d", totalNumberOfDatabasesFetched, percentageDatabasesFetched, totalNumberOfTablesFetched, percentageTablesFetched, totalNumberOfPartitionsFetched, percentagePartitionsFetched, totalNumberOfDatabases, totalNumberOfTables, totalNumberOfPartitions); LOGGER.info(snapshotFetchStatusString); printMessageTime = System.currentTimeMillis(); } } return fullSnapshot; } }
FullUpdateInitializer implements AutoCloseable { @SuppressWarnings("squid:S00112") Map<String, Collection<String>> getFullHMSSnapshot() throws Exception { List<String> allDbStr; HMSClient c = null; try (HMSClient client = clientFactory.connect()) { c = client; LOGGER.debug("Fetch all db names"); allDbStr = client.getClient().getAllDatabases(); SentryMetrics.getInstance().databaseCount.inc(allDbStr.size()); LOGGER.info("Total number of db names fetched = {}", allDbStr.size()); } catch (Exception e) { if (c != null) { c.invalidate(); } throw e; } for (String dbName : allDbStr) { results.add(threadPool.submit(new DbTask(dbName))); } Map<String, Collection<String>> fullSnapshot = new HashMap<>(); long printMessageTime = System.currentTimeMillis(); while (!results.isEmpty()) { Future<CallResult> result = results.pop(); CallResult callResult = result.get(); if (!callResult.success()) { throw callResult.getFailure(); } Map<String, Set<String>> objectMapping = callResult.getObjectMapping().getObjects(); for (Map.Entry<String, Set<String>> entry: objectMapping.entrySet()) { String key = entry.getKey(); Set<String> val = entry.getValue(); Set<String> existingSet = (Set<String>)fullSnapshot.get(key); if (existingSet == null) { fullSnapshot.put(key, val); continue; } existingSet.addAll(val); } if(System.currentTimeMillis() - printMessageTime > printSnapshotFetchTimeInterval) { long totalNumberOfDatabases = SentryMetrics.getInstance().databaseCount.getCount(); long totalNumberOfTables = SentryMetrics.getInstance().tableCount.getCount(); long totalNumberOfPartitions = SentryMetrics.getInstance().partitionCount.getCount(); double percentageDatabasesFetched = totalNumberOfDatabases > 0? ((double)totalNumberOfDatabasesFetched/totalNumberOfDatabases)*100:0; double percentageTablesFetched = totalNumberOfTables > 0? ((double)totalNumberOfTablesFetched/totalNumberOfTables)*100:0; double percentagePartitionsFetched = totalNumberOfPartitions > 0? ((double)totalNumberOfPartitionsFetched/totalNumberOfPartitions)*100:0; String snapshotFetchStatusString = String.format("Fetching full hms snapshot: databases fetched=%d (%.2f%%); " + "tables fetched=%d (%.2f%%); partitions fetched=%d (%.2f%%); total number of databases=%d; " + "total number of tables=%d total number of partitions=%d", totalNumberOfDatabasesFetched, percentageDatabasesFetched, totalNumberOfTablesFetched, percentageTablesFetched, totalNumberOfPartitionsFetched, percentagePartitionsFetched, totalNumberOfDatabases, totalNumberOfTables, totalNumberOfPartitions); LOGGER.info(snapshotFetchStatusString); printMessageTime = System.currentTimeMillis(); } } return fullSnapshot; } FullUpdateInitializer(HiveConnectionFactory clientFactory, Configuration conf); }
FullUpdateInitializer implements AutoCloseable { @SuppressWarnings("squid:S00112") Map<String, Collection<String>> getFullHMSSnapshot() throws Exception { List<String> allDbStr; HMSClient c = null; try (HMSClient client = clientFactory.connect()) { c = client; LOGGER.debug("Fetch all db names"); allDbStr = client.getClient().getAllDatabases(); SentryMetrics.getInstance().databaseCount.inc(allDbStr.size()); LOGGER.info("Total number of db names fetched = {}", allDbStr.size()); } catch (Exception e) { if (c != null) { c.invalidate(); } throw e; } for (String dbName : allDbStr) { results.add(threadPool.submit(new DbTask(dbName))); } Map<String, Collection<String>> fullSnapshot = new HashMap<>(); long printMessageTime = System.currentTimeMillis(); while (!results.isEmpty()) { Future<CallResult> result = results.pop(); CallResult callResult = result.get(); if (!callResult.success()) { throw callResult.getFailure(); } Map<String, Set<String>> objectMapping = callResult.getObjectMapping().getObjects(); for (Map.Entry<String, Set<String>> entry: objectMapping.entrySet()) { String key = entry.getKey(); Set<String> val = entry.getValue(); Set<String> existingSet = (Set<String>)fullSnapshot.get(key); if (existingSet == null) { fullSnapshot.put(key, val); continue; } existingSet.addAll(val); } if(System.currentTimeMillis() - printMessageTime > printSnapshotFetchTimeInterval) { long totalNumberOfDatabases = SentryMetrics.getInstance().databaseCount.getCount(); long totalNumberOfTables = SentryMetrics.getInstance().tableCount.getCount(); long totalNumberOfPartitions = SentryMetrics.getInstance().partitionCount.getCount(); double percentageDatabasesFetched = totalNumberOfDatabases > 0? ((double)totalNumberOfDatabasesFetched/totalNumberOfDatabases)*100:0; double percentageTablesFetched = totalNumberOfTables > 0? ((double)totalNumberOfTablesFetched/totalNumberOfTables)*100:0; double percentagePartitionsFetched = totalNumberOfPartitions > 0? ((double)totalNumberOfPartitionsFetched/totalNumberOfPartitions)*100:0; String snapshotFetchStatusString = String.format("Fetching full hms snapshot: databases fetched=%d (%.2f%%); " + "tables fetched=%d (%.2f%%); partitions fetched=%d (%.2f%%); total number of databases=%d; " + "total number of tables=%d total number of partitions=%d", totalNumberOfDatabasesFetched, percentageDatabasesFetched, totalNumberOfTablesFetched, percentageTablesFetched, totalNumberOfPartitionsFetched, percentagePartitionsFetched, totalNumberOfDatabases, totalNumberOfTables, totalNumberOfPartitions); LOGGER.info(snapshotFetchStatusString); printMessageTime = System.currentTimeMillis(); } } return fullSnapshot; } FullUpdateInitializer(HiveConnectionFactory clientFactory, Configuration conf); @Override void close(); }
FullUpdateInitializer implements AutoCloseable { @SuppressWarnings("squid:S00112") Map<String, Collection<String>> getFullHMSSnapshot() throws Exception { List<String> allDbStr; HMSClient c = null; try (HMSClient client = clientFactory.connect()) { c = client; LOGGER.debug("Fetch all db names"); allDbStr = client.getClient().getAllDatabases(); SentryMetrics.getInstance().databaseCount.inc(allDbStr.size()); LOGGER.info("Total number of db names fetched = {}", allDbStr.size()); } catch (Exception e) { if (c != null) { c.invalidate(); } throw e; } for (String dbName : allDbStr) { results.add(threadPool.submit(new DbTask(dbName))); } Map<String, Collection<String>> fullSnapshot = new HashMap<>(); long printMessageTime = System.currentTimeMillis(); while (!results.isEmpty()) { Future<CallResult> result = results.pop(); CallResult callResult = result.get(); if (!callResult.success()) { throw callResult.getFailure(); } Map<String, Set<String>> objectMapping = callResult.getObjectMapping().getObjects(); for (Map.Entry<String, Set<String>> entry: objectMapping.entrySet()) { String key = entry.getKey(); Set<String> val = entry.getValue(); Set<String> existingSet = (Set<String>)fullSnapshot.get(key); if (existingSet == null) { fullSnapshot.put(key, val); continue; } existingSet.addAll(val); } if(System.currentTimeMillis() - printMessageTime > printSnapshotFetchTimeInterval) { long totalNumberOfDatabases = SentryMetrics.getInstance().databaseCount.getCount(); long totalNumberOfTables = SentryMetrics.getInstance().tableCount.getCount(); long totalNumberOfPartitions = SentryMetrics.getInstance().partitionCount.getCount(); double percentageDatabasesFetched = totalNumberOfDatabases > 0? ((double)totalNumberOfDatabasesFetched/totalNumberOfDatabases)*100:0; double percentageTablesFetched = totalNumberOfTables > 0? ((double)totalNumberOfTablesFetched/totalNumberOfTables)*100:0; double percentagePartitionsFetched = totalNumberOfPartitions > 0? ((double)totalNumberOfPartitionsFetched/totalNumberOfPartitions)*100:0; String snapshotFetchStatusString = String.format("Fetching full hms snapshot: databases fetched=%d (%.2f%%); " + "tables fetched=%d (%.2f%%); partitions fetched=%d (%.2f%%); total number of databases=%d; " + "total number of tables=%d total number of partitions=%d", totalNumberOfDatabasesFetched, percentageDatabasesFetched, totalNumberOfTablesFetched, percentageTablesFetched, totalNumberOfPartitionsFetched, percentagePartitionsFetched, totalNumberOfDatabases, totalNumberOfTables, totalNumberOfPartitions); LOGGER.info(snapshotFetchStatusString); printMessageTime = System.currentTimeMillis(); } } return fullSnapshot; } FullUpdateInitializer(HiveConnectionFactory clientFactory, Configuration conf); @Override void close();  }