CachedHttpFileSystem extends FileSystem { @Override public FSDataInputStream open(final Path path, final int bufferSize) throws IOException { URI relativeURI = this.uri.relativize(path.toUri()); if (relativeURI.isAbsolute()) { throw new IOException("Path must be relative or have same prefix as root origin URI."); } URI resolvedOriginURI = this.rootOriginURI.resolve(relativeURI); if (!this.cachingEnabled) { this.log.info("CACHE MISS (cache is disabled): " + resolvedOriginURI.toString()); return downloadFromOrigin(resolvedOriginURI); } Path resolvedCachePath = new Path(this.rootCachedURI.resolve(relativeURI)); try { FSDataInputStream cachedInputStream = this.cacheFS.open(resolvedCachePath, bufferSize); this.log.info("CACHE HIT: " + resolvedCachePath.toString()); return cachedInputStream; } catch (FileNotFoundException e) { this.log.info("CACHE MISS: " + resolvedCachePath.toString()); FSDataInputStream originInputStream = downloadFromOrigin(resolvedOriginURI); Path folderInCacheForFile = resolvedCachePath.getParent(); Path tempCacheFile = new Path(folderInCacheForFile, String.format(CACHE_TMP_FILE_TEMPLATE, RAND.nextInt())); FSDataOutputStream outStreamToTmpFile; try { this.cacheFS.mkdirs(folderInCacheForFile); outStreamToTmpFile = this.cacheFS.create(tempCacheFile, false); } catch (IOException e2) { this.log.warn("Failed to persist file to cache, returning stream from origin: " + resolvedOriginURI.toString(), e2); return originInputStream; } IOUtils.copy(originInputStream, outStreamToTmpFile); originInputStream.close(); outStreamToTmpFile.close(); try { this.cacheFS.rename(tempCacheFile, resolvedCachePath); } catch (FileAlreadyExistsException e2) { this.log.info("Another process already persisted this file: " + resolvedCachePath.toString()); } return this.cacheFS.open(resolvedCachePath, bufferSize); } }  @Override void initialize(final URI name, final Configuration conf); @Override String getScheme(); @Override URI getUri(); @Override FSDataInputStream open(final Path path, final int bufferSize); @Override FSDataOutputStream create(final Path path, final FsPermission fsPermission,
      final boolean b, final int i, final short i1, final long l,
      final Progressable progressable); @Override FSDataOutputStream append(final Path path, final int i, final Progressable progressable); @Override boolean rename(final Path path, final Path path1); @Override boolean delete(final Path path, final boolean b); @Override FileStatus[] listStatus(final Path path); @Override void setWorkingDirectory(final Path path); @Override Path getWorkingDirectory(); @Override boolean mkdirs(final Path path, final FsPermission fsPermission); @Override FileStatus getFileStatus(final Path path); static final String CACHE_ROOT_URI; static final String CACHE_ENABLED_FLAG; }
@Test public void testCachingDisabled() throws Exception { Configuration conf = new Configuration(false); conf.setBoolean("fs.chttp.impl.disable.cache", true); conf.set("fs.chttp.impl", azkaban.cachedhttpfilesystem.CachedHttpFileSystem.class.getName()); conf.set(CachedHttpFileSystem.CACHE_ENABLED_FLAG, "false"); CachedHttpFileSystem tmpCachedHttpFileSystem = spy((CachedHttpFileSystem) FileSystem.get(chttpRootURI, conf)); configureMockResponseFromHttpOrigin(tmpCachedHttpFileSystem, this.originAbsoluteJarURI); assertInputStreamIsJarContent(tmpCachedHttpFileSystem.open(new Path(this.chttpAbsoluteJarURI))); }
@Test(expected = IOException.class) public void testInvalidUriPath() throws Exception { this.cachedHttpFileSystem.open(new Path("chttp: }
@Test(expected = IOException.class) public void testInvalidUriScheme() throws Exception { this.cachedHttpFileSystem.open(new Path("http: }