AllocationFileLoaderService extends AbstractService { @VisibleForTesting Path getAllocationFile(Configuration conf) throws UnsupportedFileSystemException { String allocFilePath = conf.get(FairSchedulerConfiguration.ALLOCATION_FILE, FairSchedulerConfiguration.DEFAULT_ALLOCATION_FILE); Path allocPath = new Path(allocFilePath); String allocPathScheme = allocPath.toUri().getScheme(); if(allocPathScheme != null && !allocPathScheme.matches(SUPPORTED_FS_REGEX)){ throw new UnsupportedFileSystemException("Allocation file " + allocFilePath + " uses an unsupported filesystem"); } else if (!allocPath.isAbsolute()) { URL url = Thread.currentThread().getContextClassLoader() .getResource(allocFilePath); if (url == null) { LOG.warn(allocFilePath + " not found on the classpath."); allocPath = null; } else if (!url.getProtocol().equalsIgnoreCase("file")) { throw new RuntimeException("Allocation file " + url + " found on the classpath is not on the local filesystem."); } else { allocPath = new Path(url.getProtocol(), null, url.getPath()); } } else if (allocPath.isAbsoluteAndSchemeAuthorityNull()){ allocPath = new Path("file", null, allocFilePath); } return allocPath; } AllocationFileLoaderService();  AllocationFileLoaderService(Clock clock); @Override void serviceInit(Configuration conf); @Override void serviceStart(); @Override void serviceStop(); synchronized void setReloadListener(Listener reloadListener); synchronized void reloadAllocations(); static final Log LOG; static final long ALLOC_RELOAD_INTERVAL_MS; static final long ALLOC_RELOAD_WAIT_MS; static final long THREAD_JOIN_TIMEOUT_MS; }
@Test public void testGetAllocationFileFromFileSystem() throws IOException, URISyntaxException { Configuration conf = new YarnConfiguration(); File baseDir = new File(TEST_DIR + Path.SEPARATOR + "getAllocHDFS").getAbsoluteFile(); FileUtil.fullyDelete(baseDir); conf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, baseDir.getAbsolutePath()); MiniDFSCluster.Builder builder = new MiniDFSCluster.Builder(conf); MiniDFSCluster hdfsCluster = builder.build(); String fsAllocPath = "hdfs: + Path.SEPARATOR + TEST_FAIRSCHED_XML; URL fschedURL = Thread.currentThread().getContextClassLoader() .getResource(TEST_FAIRSCHED_XML); FileSystem fs = FileSystem.get(conf); fs.copyFromLocalFile(new Path(fschedURL.toURI()), new Path(fsAllocPath)); conf.set(FairSchedulerConfiguration.ALLOCATION_FILE, fsAllocPath); AllocationFileLoaderService allocLoader = new AllocationFileLoaderService(); Path allocationFile = allocLoader.getAllocationFile(conf); assertEquals(fsAllocPath, allocationFile.toString()); assertTrue(fs.exists(allocationFile)); hdfsCluster.shutdown(true); }
@Test (expected = UnsupportedFileSystemException.class) public void testDenyGetAllocationFileFromUnsupportedFileSystem() throws UnsupportedFileSystemException { Configuration conf = new YarnConfiguration(); conf.set(FairSchedulerConfiguration.ALLOCATION_FILE, "badfs: AllocationFileLoaderService allocLoader = new AllocationFileLoaderService(); allocLoader.getAllocationFile(conf); }
@Test public void testGetAllocationFileFromClasspath() { try { Configuration conf = new Configuration(); FileSystem fs = FileSystem.get(conf); conf.set(FairSchedulerConfiguration.ALLOCATION_FILE, TEST_FAIRSCHED_XML); AllocationFileLoaderService allocLoader = new AllocationFileLoaderService(); Path allocationFile = allocLoader.getAllocationFile(conf); assertEquals(TEST_FAIRSCHED_XML, allocationFile.getName()); assertTrue(fs.exists(allocationFile)); } catch (IOException e) { fail("Unable to access allocation file from classpath: " + e); } }