@Test public void testExceptionInTask() throws Exception { Database db1 = Mockito.mock(Database.class); Mockito.when(db1.getName()).thenReturn("db1"); Mockito.when(db1.getLocationUri()).thenReturn("hdfs: Table tab1 = Mockito.mock(Table.class); Mockito.when(tab1.getDbName()).thenReturn("db2"); Mockito.when(tab1.getTableName()).thenReturn("tab1"); IHMSHandler hmsHandler = Mockito.mock(IHMSHandler.class); Mockito.when(hmsHandler.get_all_databases()).thenReturn(Lists .newArrayList("db1")); Mockito.when(hmsHandler.get_database("db1")).thenReturn(db1); Mockito.when(hmsHandler.get_table_objects_by_name("db1", Lists.newArrayList("tab1"))) .thenReturn(Lists.newArrayList(tab1)); Mockito.when(hmsHandler.get_all_tables("db1")).thenReturn(Lists .newArrayList("tab1")); Configuration conf = new Configuration(); conf.setInt(ServiceConstants.ServerConfig .SENTRY_HDFS_SYNC_METASTORE_CACHE_MAX_PART_PER_RPC, 1); conf.setInt(ServiceConstants.ServerConfig .SENTRY_HDFS_SYNC_METASTORE_CACHE_MAX_TABLES_PER_RPC, 1); conf.setInt(ServiceConstants.ServerConfig .SENTRY_HDFS_SYNC_METASTORE_CACHE_INIT_THREADS, 1); conf.setInt(ServiceConstants.ServerConfig .SENTRY_HDFS_SYNC_METASTORE_CACHE_RETRY_MAX_NUM, 2); try { MetastoreCacheInitializer cacheInitializer = new MetastoreCacheInitializer(hmsHandler, conf); cacheInitializer.createInitialUpdate(); Assert.fail("Expected cacheInitializer to fail"); } catch (Exception e) { Assert.assertTrue(e instanceof RuntimeException); } }
UpdateableAuthzPaths createInitialUpdate() throws Exception { UpdateableAuthzPaths authzPaths = new UpdateableAuthzPaths(new String[]{"/"}); PathsUpdate tempUpdate = new PathsUpdate(-1, false); List<String> allDbStr = hmsHandler.get_all_databases(); for (String dbName : allDbStr) { Callable<CallResult> dbTask = new DbTask(tempUpdate, dbName); results.add(threadPool.submit(dbTask)); } while (taskCounter.get() > 0) { Thread.sleep(1000); } for (Future<CallResult> result : results) { CallResult callResult = result.get(); if (callResult.getSuccessStatus() == false && failOnRetry) { throw new RuntimeException(callResult.getFailure()); } } authzPaths.updatePartial(Lists.newArrayList(tempUpdate), new ReentrantReadWriteLock()); return authzPaths; }
MetastoreCacheInitializer implements Closeable { UpdateableAuthzPaths createInitialUpdate() throws Exception { UpdateableAuthzPaths authzPaths = new UpdateableAuthzPaths(new String[]{"/"}); PathsUpdate tempUpdate = new PathsUpdate(-1, false); List<String> allDbStr = hmsHandler.get_all_databases(); for (String dbName : allDbStr) { Callable<CallResult> dbTask = new DbTask(tempUpdate, dbName); results.add(threadPool.submit(dbTask)); } while (taskCounter.get() > 0) { Thread.sleep(1000); } for (Future<CallResult> result : results) { CallResult callResult = result.get(); if (callResult.getSuccessStatus() == false && failOnRetry) { throw new RuntimeException(callResult.getFailure()); } } authzPaths.updatePartial(Lists.newArrayList(tempUpdate), new ReentrantReadWriteLock()); return authzPaths; } }
MetastoreCacheInitializer implements Closeable { UpdateableAuthzPaths createInitialUpdate() throws Exception { UpdateableAuthzPaths authzPaths = new UpdateableAuthzPaths(new String[]{"/"}); PathsUpdate tempUpdate = new PathsUpdate(-1, false); List<String> allDbStr = hmsHandler.get_all_databases(); for (String dbName : allDbStr) { Callable<CallResult> dbTask = new DbTask(tempUpdate, dbName); results.add(threadPool.submit(dbTask)); } while (taskCounter.get() > 0) { Thread.sleep(1000); } for (Future<CallResult> result : results) { CallResult callResult = result.get(); if (callResult.getSuccessStatus() == false && failOnRetry) { throw new RuntimeException(callResult.getFailure()); } } authzPaths.updatePartial(Lists.newArrayList(tempUpdate), new ReentrantReadWriteLock()); return authzPaths; } MetastoreCacheInitializer(IHMSHandler hmsHandler, Configuration conf); }
MetastoreCacheInitializer implements Closeable { UpdateableAuthzPaths createInitialUpdate() throws Exception { UpdateableAuthzPaths authzPaths = new UpdateableAuthzPaths(new String[]{"/"}); PathsUpdate tempUpdate = new PathsUpdate(-1, false); List<String> allDbStr = hmsHandler.get_all_databases(); for (String dbName : allDbStr) { Callable<CallResult> dbTask = new DbTask(tempUpdate, dbName); results.add(threadPool.submit(dbTask)); } while (taskCounter.get() > 0) { Thread.sleep(1000); } for (Future<CallResult> result : results) { CallResult callResult = result.get(); if (callResult.getSuccessStatus() == false && failOnRetry) { throw new RuntimeException(callResult.getFailure()); } } authzPaths.updatePartial(Lists.newArrayList(tempUpdate), new ReentrantReadWriteLock()); return authzPaths; } MetastoreCacheInitializer(IHMSHandler hmsHandler, Configuration conf); @Override void close(); }
MetastoreCacheInitializer implements Closeable { UpdateableAuthzPaths createInitialUpdate() throws Exception { UpdateableAuthzPaths authzPaths = new UpdateableAuthzPaths(new String[]{"/"}); PathsUpdate tempUpdate = new PathsUpdate(-1, false); List<String> allDbStr = hmsHandler.get_all_databases(); for (String dbName : allDbStr) { Callable<CallResult> dbTask = new DbTask(tempUpdate, dbName); results.add(threadPool.submit(dbTask)); } while (taskCounter.get() > 0) { Thread.sleep(1000); } for (Future<CallResult> result : results) { CallResult callResult = result.get(); if (callResult.getSuccessStatus() == false && failOnRetry) { throw new RuntimeException(callResult.getFailure()); } } authzPaths.updatePartial(Lists.newArrayList(tempUpdate), new ReentrantReadWriteLock()); return authzPaths; } MetastoreCacheInitializer(IHMSHandler hmsHandler, Configuration conf); @Override void close();  }