UploadStagingCache implements Closeable { @Override public void close() { LOG.info("Uploads in progress on close [{}]", map.size()); LOG.info("Uploads completed but not cleared from cache [{}]", attic.size()); LOG.info("Staging cache stats on close [{}]", cacheStats.cacheInfoAsString()); new ExecutorCloser(executor).close(); new ExecutorCloser(scheduledExecutor).close(); new ExecutorCloser(statsExecutor).close(); } private  UploadStagingCache(File dir, File home, int uploadThreads, long size /* bytes */,
        StagingUploader uploader, @Nullable FileCache cache, StatisticsProvider statisticsProvider,
        @Nullable ListeningExecutorService executor,
        @Nullable ScheduledExecutorService scheduledExecutor,
        int purgeInterval /* secs */, int retryInterval /* secs */); private  UploadStagingCache(); static UploadStagingCache build(File dir, File home, int uploadThreads, long size
        /* bytes */, StagingUploader uploader, @Nullable FileCache cache,
        StatisticsProvider statisticsProvider, @Nullable ListeningExecutorService executor,
        @Nullable ScheduledExecutorService scheduledExecutor, int purgeInterval /* secs */,
        int retryInterval /* secs */); Optional<SettableFuture<Integer>> put(String id, File input); @Nullable File getIfPresent(String key); DataStoreCacheStatsMBean getStats(); @Override void close();  }
@Test public void testConcurrentSameAddRequest() throws Exception { LOG.info("Starting testConcurrentSameAddRequest"); closer.close(); ListeningExecutorService executorService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(2)); List<ListenableFuture<Integer>> futures = Lists.newArrayList(); CountDownLatch moveLatch = new CountDownLatch(1); init(1, new TestStagingUploader(folder.newFolder(), moveLatch), null); ListenableFuture<Boolean> resultReq1 = putThread(folder, executorService, futures); Thread.sleep(100); ListenableFuture<Boolean> resultReq2 = putThread(folder, executorService, futures); Thread.sleep(200); moveLatch.countDown(); assertTrue(resultReq1.get()); assertTrue(resultReq2.get()); taskLatch.countDown(); callbackLatch.countDown(); assertFuture(futures, 0); assertCacheStats(stagingCache, 0, 0, 1, 2); LOG.info("Finished testConcurrentSameAddRequest"); }