@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);
    }
  }
@Test
  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);

    // the FileSystem object owned by SystemServiceManager must not be closed
    // when cleanup a service
    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();
      }
    }
  }