SystemServiceManagerImpl extends AbstractService implements SystemServiceManager { @Override protected void serviceInit(Configuration conf) throws Exception { String dirPath = conf.get(YarnServiceConf.YARN_SERVICES_SYSTEM_SERVICE_DIRECTORY); if (dirPath != null) { systemServiceDir = new Path(dirPath); LOG.info("System Service Directory is configured to {}", systemServiceDir); fs = systemServiceDir.getFileSystem(conf); this.loginUGI = UserGroupInformation.isSecurityEnabled() ? UserGroupInformation.getLoginUser() : UserGroupInformation.getCurrentUser(); LOG.info("UserGroupInformation initialized to {}", loginUGI); } } SystemServiceManagerImpl();   }
@Test public void testFileSystemCloseWhenCleanUpService() throws Exception { FileSystem fs = null; Path path = new Path("/tmp/servicedir"); HdfsConfiguration hdfsConfig = new HdfsConfiguration(); MiniDFSCluster hdfsCluster = new MiniDFSCluster.Builder(hdfsConfig) .numDataNodes(1).build(); fs = hdfsCluster.getFileSystem(); if (!fs.exists(path)) { fs.mkdirs(path); } SystemServiceManagerImpl serviceManager = new SystemServiceManagerImpl(); hdfsConfig.set(YarnServiceConf.YARN_SERVICES_SYSTEM_SERVICE_DIRECTORY, path.toString()); serviceManager.init(hdfsConfig); hdfsConfig.set("hadoop.registry.zk.connection.timeout.ms", "100"); hdfsConfig.set("hadoop.registry.zk.retry.times", "1"); ApiServiceClient asc = new ApiServiceClient(); asc.serviceInit(hdfsConfig); asc.actionCleanUp("testapp", "testuser"); try { serviceManager.start(); } catch (Exception e) { if (e.getMessage().contains("Filesystem closed")) { fail("SystemServiceManagerImpl failed to handle " + "FileSystem close"); } else { fail("Should not get any exceptions"); } } finally { serviceManager.stop(); fs = hdfsCluster.getFileSystem(); if (fs.exists(path)) { fs.delete(path, true); } if (hdfsCluster != null) { hdfsCluster.shutdown(); } } }