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();  }
@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")); }
@Test public void testInvalidPaths() throws Exception { Database db1 = makeDb("db1"); Table tab1 = Mockito.mock(Table.class); Mockito.when(tab1.getDbName()).thenReturn("db2"); Mockito.when(tab1.getTableName()).thenReturn("tab1"); HiveMetaStoreClient client = Mockito.mock(HiveMetaStoreClient.class); Mockito.when(client.getAllDatabases()).thenReturn(Lists.newArrayList("db1")); Mockito.when(client.getDatabase("db1")).thenReturn(db1); Table tab12 = Mockito.mock(Table.class); Mockito.when(tab12.getDbName()).thenReturn("db1"); Mockito.when(tab12.getTableName()).thenReturn("tab21"); StorageDescriptor sd21 = Mockito.mock(StorageDescriptor.class); Mockito.when(sd21.getLocation()).thenReturn("hdfs: Mockito.when(tab12.getSd()).thenReturn(sd21); Mockito.when(client.getTableObjectsByName("db1", Lists.newArrayList("tab1"))).thenReturn(Lists.newArrayList(tab1)); Mockito.when(client.getTableObjectsByName("db1", Lists.newArrayList("tab12"))).thenReturn(Lists.newArrayList(tab12)); Mockito.when(client.getAllTables("db1")). thenReturn(Lists.newArrayList("tab1", "tab12")); Map<String, Collection<String>> update; try(FullUpdateInitializer cacheInitializer = new FullUpdateInitializer(new MockHMSClientFactory(client), conf)) { update = cacheInitializer.getFullHMSSnapshot(); } Assert.assertEquals(2, update.size()); Assert.assertEquals(Sets.newHashSet("db1"), update.get("db1")); Assert.assertEquals(Sets.newHashSet("db1/tab21"), update.get("db1.tab21")); }
@Test public void testBig() throws Exception { int ndbs = 3; int ntables = 51; int nparts = 131; HiveSnapshot snap = new HiveSnapshot(); for (int i = 0; i < ndbs; i++) { HiveDb db = new HiveDb("db" + i); for (int j = 0; j < ntables; j++) { HiveTable table = new HiveTable("table" + i + j); for (int k = 0; k < nparts; k++) { table.add("part" + i + j + k); } db.add(table); } snap.add(db); } MockClient c = new MockClient(snap); Map<String, Collection<String>> update; try(FullUpdateInitializer cacheInitializer = new FullUpdateInitializer(new MockHMSClientFactory(c), conf)) { update = cacheInitializer.getFullHMSSnapshot(); } Assert.assertEquals((ntables * ndbs) + ndbs, update.size()); for (int i = 0; i < ndbs; i++) { String dbName = "db" + i; Assert.assertEquals(Sets.newHashSet(dbName), update.get(dbName)); for (int j = 0; j < ntables; j++) { String tableName = "table" + i + j; Set<String> values = new HashSet<>(); values.add(String.format("%s/%s", dbName, tableName)); for (int k = 0; k < nparts; k++) { String partName = "part" + i + j + k; values.add(String.format("%s/%s/%s", dbName, tableName, partName)); } String authz = dbName + "." + tableName; Assert.assertEquals(values, update.get(authz)); } } }