@Test public void testUseFallback() throws Exception { createUfsDirs("a"); createUfsFile("a/b"); Collection<UfsStatus> children = mCache .fetchChildrenIfAbsent(new AlluxioURI("/a"), mMountTable); assertEquals(1, children.size()); children.forEach(stat -> assertEquals("b", stat.getName())); }
@Nullable public Collection<UfsStatus> fetchChildrenIfAbsent(AlluxioURI path, MountTable mountTable, boolean useFallback) throws InterruptedException, InvalidPathException { Future<Collection<UfsStatus>> prefetchJob = mActivePrefetchJobs.get(path); if (prefetchJob != null) { try { return prefetchJob.get(); } catch (InterruptedException | ExecutionException e) { LogUtils.warnWithException(LOG, "Failed to get result for prefetch job on alluxio path {}", path, e); if (e instanceof InterruptedException) { Thread.currentThread().interrupt(); throw (InterruptedException) e; } } finally { mActivePrefetchJobs.remove(path); } } Collection<UfsStatus> children = getChildren(path); if (children != null) { return children; } if (useFallback) { return getChildrenIfAbsent(path, mountTable); } return null; }
UfsStatusCache { @Nullable public Collection<UfsStatus> fetchChildrenIfAbsent(AlluxioURI path, MountTable mountTable, boolean useFallback) throws InterruptedException, InvalidPathException { Future<Collection<UfsStatus>> prefetchJob = mActivePrefetchJobs.get(path); if (prefetchJob != null) { try { return prefetchJob.get(); } catch (InterruptedException | ExecutionException e) { LogUtils.warnWithException(LOG, "Failed to get result for prefetch job on alluxio path {}", path, e); if (e instanceof InterruptedException) { Thread.currentThread().interrupt(); throw (InterruptedException) e; } } finally { mActivePrefetchJobs.remove(path); } } Collection<UfsStatus> children = getChildren(path); if (children != null) { return children; } if (useFallback) { return getChildrenIfAbsent(path, mountTable); } return null; } }
UfsStatusCache { @Nullable public Collection<UfsStatus> fetchChildrenIfAbsent(AlluxioURI path, MountTable mountTable, boolean useFallback) throws InterruptedException, InvalidPathException { Future<Collection<UfsStatus>> prefetchJob = mActivePrefetchJobs.get(path); if (prefetchJob != null) { try { return prefetchJob.get(); } catch (InterruptedException | ExecutionException e) { LogUtils.warnWithException(LOG, "Failed to get result for prefetch job on alluxio path {}", path, e); if (e instanceof InterruptedException) { Thread.currentThread().interrupt(); throw (InterruptedException) e; } } finally { mActivePrefetchJobs.remove(path); } } Collection<UfsStatus> children = getChildren(path); if (children != null) { return children; } if (useFallback) { return getChildrenIfAbsent(path, mountTable); } return null; } UfsStatusCache(@Nullable ExecutorService prefetchExecutor); }
UfsStatusCache { @Nullable public Collection<UfsStatus> fetchChildrenIfAbsent(AlluxioURI path, MountTable mountTable, boolean useFallback) throws InterruptedException, InvalidPathException { Future<Collection<UfsStatus>> prefetchJob = mActivePrefetchJobs.get(path); if (prefetchJob != null) { try { return prefetchJob.get(); } catch (InterruptedException | ExecutionException e) { LogUtils.warnWithException(LOG, "Failed to get result for prefetch job on alluxio path {}", path, e); if (e instanceof InterruptedException) { Thread.currentThread().interrupt(); throw (InterruptedException) e; } } finally { mActivePrefetchJobs.remove(path); } } Collection<UfsStatus> children = getChildren(path); if (children != null) { return children; } if (useFallback) { return getChildrenIfAbsent(path, mountTable); } return null; } UfsStatusCache(@Nullable ExecutorService prefetchExecutor); @Nullable UfsStatus addStatus(AlluxioURI path, UfsStatus status); @Nullable Collection<UfsStatus> addChildren(AlluxioURI path, Collection<UfsStatus> children); @Nullable UfsStatus remove(AlluxioURI path); @Nullable UfsStatus getStatus(AlluxioURI path); @Nullable UfsStatus fetchStatusIfAbsent(AlluxioURI path, MountTable mountTable); @Nullable Collection<UfsStatus> fetchChildrenIfAbsent(AlluxioURI path, MountTable mountTable,
      boolean useFallback); @Nullable Collection<UfsStatus> fetchChildrenIfAbsent(AlluxioURI path, MountTable mountTable); @Nullable Collection<UfsStatus> getChildren(AlluxioURI path); @Nullable Future<Collection<UfsStatus>> prefetchChildren(AlluxioURI path, MountTable mountTable); void cancelAllPrefetch(); }
UfsStatusCache { @Nullable public Collection<UfsStatus> fetchChildrenIfAbsent(AlluxioURI path, MountTable mountTable, boolean useFallback) throws InterruptedException, InvalidPathException { Future<Collection<UfsStatus>> prefetchJob = mActivePrefetchJobs.get(path); if (prefetchJob != null) { try { return prefetchJob.get(); } catch (InterruptedException | ExecutionException e) { LogUtils.warnWithException(LOG, "Failed to get result for prefetch job on alluxio path {}", path, e); if (e instanceof InterruptedException) { Thread.currentThread().interrupt(); throw (InterruptedException) e; } } finally { mActivePrefetchJobs.remove(path); } } Collection<UfsStatus> children = getChildren(path); if (children != null) { return children; } if (useFallback) { return getChildrenIfAbsent(path, mountTable); } return null; } UfsStatusCache(@Nullable ExecutorService prefetchExecutor); @Nullable UfsStatus addStatus(AlluxioURI path, UfsStatus status); @Nullable Collection<UfsStatus> addChildren(AlluxioURI path, Collection<UfsStatus> children); @Nullable UfsStatus remove(AlluxioURI path); @Nullable UfsStatus getStatus(AlluxioURI path); @Nullable UfsStatus fetchStatusIfAbsent(AlluxioURI path, MountTable mountTable); @Nullable Collection<UfsStatus> fetchChildrenIfAbsent(AlluxioURI path, MountTable mountTable,
      boolean useFallback); @Nullable Collection<UfsStatus> fetchChildrenIfAbsent(AlluxioURI path, MountTable mountTable); @Nullable Collection<UfsStatus> getChildren(AlluxioURI path); @Nullable Future<Collection<UfsStatus>> prefetchChildren(AlluxioURI path, MountTable mountTable); void cancelAllPrefetch();  }