FileCache extends AbstractCache<String, File> implements Closeable { public File get(String key) throws IOException { try { return cache.get(key, () -> cacheLoader.load(key)); } catch (ExecutionException e) { LOG.error("Error loading [{}] from cache", key); throw new IOException(e); } } private  FileCache(long maxSize /* bytes */, File root,
        final CacheLoader<String, InputStream> loader, @Nullable final ExecutorService executor); private  FileCache(); static FileCache build(long maxSize /* bytes */, File root,
        final CacheLoader<String, InputStream> loader, @Nullable final ExecutorService executor); @Override void put(String key, File file); boolean containsKey(String key); @Nullable File getIfPresent(String key); @Nullable @Override File getIfPresent(Object key); File get(String key); @Override void invalidate(Object key); DataStoreCacheStatsMBean getStats(); @Override void close();  }
@Test public void retrieveSameConcurrent() throws Exception { LOG.info("Started retrieveSameConcurrent"); File f = createFile(0, loader, cache, folder); ListeningExecutorService executorService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(2)); closer.register(new ExecutorCloser(executorService, 5, TimeUnit.MILLISECONDS)); CountDownLatch thread1Start = new CountDownLatch(1); SettableFuture<File> future1 = retrieveThread(executorService, ID_PREFIX + 0, cache, thread1Start); CountDownLatch thread2Start = new CountDownLatch(1); SettableFuture<File> future2 = retrieveThread(executorService, ID_PREFIX + 0, cache, thread2Start); thread1Start.countDown(); thread2Start.countDown(); future1.get(); future2.get(); LOG.info("Async tasks finished"); assertCacheIfPresent(0, cache, f); assertCacheStats(cache, 1, 4 * 1024, 1, 1); LOG.info("Finished retrieveSameConcurrent"); }
@Test public void retrievePutConcurrent() throws Exception { LOG.info("Started retrievePutConcurrent"); final File f = createFile(0, loader, cache, folder); File f2 = copyToFile(randomStream(1, 4 * 1024), folder.newFile()); ListeningExecutorService executorService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(2)); closer.register(new ExecutorCloser(executorService, 5, TimeUnit.MILLISECONDS)); CountDownLatch thread1Start = new CountDownLatch(1); SettableFuture<File> future1 = retrieveThread(executorService, ID_PREFIX + 0, cache, thread1Start); CountDownLatch thread2Start = new CountDownLatch(1); SettableFuture<Boolean> future2 = putThread(executorService, 1, f2, cache, thread2Start); thread1Start.countDown(); thread2Start.countDown(); future1.get(); future2.get(); LOG.info("Async tasks finished"); assertCacheIfPresent(0, cache, f); assertCacheIfPresent(1, cache, copyToFile(randomStream(1, 4 * 1024), folder.newFile())); assertCacheStats(cache, 2, 8 * 1024, 1, 1); LOG.info("Finished retrievePutConcurrent"); }
@Test public void getInvalidateConcurrent() throws Exception { LOG.info("Started getInvalidateConcurrent"); for (int i = 0; i < 15; i++) { if (i != 4) { File f = createFile(i, loader, cache, folder); assertCache(i, cache, f); } } LOG.info("Finished creating load"); ListeningExecutorService executorService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(2)); closer.register(new ExecutorCloser(executorService, 5, TimeUnit.MILLISECONDS)); CountDownLatch thread1Start = new CountDownLatch(1); SettableFuture<File> future1 = retrieveThread(executorService, ID_PREFIX + 10, cache, thread1Start); thread1Start.countDown(); File f = createFile(4, loader, cache, folder); CountDownLatch thread2Start = new CountDownLatch(1); SettableFuture<File> future2 = retrieveThread(executorService, ID_PREFIX + 4, cache, thread2Start); thread2Start.countDown(); File f10 = future1.get(); File f4 = future2.get(); LOG.info("Async tasks finished"); if (f10.exists()) { assertCacheIfPresent(10, cache, f10); } if (f4.exists()) { assertCacheIfPresent(4, cache, f4); } LOG.info("Finished getInvalidateConcurrent"); }