@Test public void conflict() throws Exception { File source = folder.newFile(); byte[] bytes = new byte[1024 * 1024]; try (OutputStream output = new FileOutputStream(source)) { for (int i = 0, n = 50; i < n; i++) { output.write(bytes); } } Path path = path(source); File cacheRepo = folder.newFolder(); Configuration configuration = new ConfigurationProvider().newInstance(); LockProvider<Path> locks = new LocalFileLockProvider<>(folder.newFolder()); RetryStrategy retrier = new ConstantRetryStrategy(30, 100, 200); FileCacheRepository cache = new HadoopFileCacheRepository(configuration, path(cacheRepo), locks, retrier); List<Future<Path>> futures = new ArrayList<>(); int count = 10; CountDownLatch latch = new CountDownLatch(count); ExecutorService executor = Executors.newFixedThreadPool(count); try { for (int i = 0; i < count; i++) { String label = String.format("thread-%d", i); futures.add(executor.submit(() -> { LOG.info("Wait: resolve @" + label); latch.countDown(); if (latch.await(5, TimeUnit.SECONDS) == false) { throw new TimeoutException(); } LOG.info("Start: resolve @" + label); Path result = cache.resolve(path); LOG.info("Finish: resolve @" + label); return result; })); } executor.shutdown(); if (executor.awaitTermination(30, TimeUnit.SECONDS) == false) { throw new TimeoutException(); } } finally { executor.shutdownNow(); } for (Future<Path> future : futures) { future.get(); } }
@Override public Path resolve(Path file) throws IOException, InterruptedException { FileSystem fs = file.getFileSystem(configuration); Path qualified = fs.makeQualified(file); return doResolve(qualified); }
HadoopFileCacheRepository implements FileCacheRepository { @Override public Path resolve(Path file) throws IOException, InterruptedException { FileSystem fs = file.getFileSystem(configuration); Path qualified = fs.makeQualified(file); return doResolve(qualified); } }
HadoopFileCacheRepository implements FileCacheRepository { @Override public Path resolve(Path file) throws IOException, InterruptedException { FileSystem fs = file.getFileSystem(configuration); Path qualified = fs.makeQualified(file); return doResolve(qualified); } HadoopFileCacheRepository(
            Configuration configuration,
            Path repository,
            LockProvider<? super Path> lockProvider,
            RetryStrategy retryStrategy); }
HadoopFileCacheRepository implements FileCacheRepository { @Override public Path resolve(Path file) throws IOException, InterruptedException { FileSystem fs = file.getFileSystem(configuration); Path qualified = fs.makeQualified(file); return doResolve(qualified); } HadoopFileCacheRepository(
            Configuration configuration,
            Path repository,
            LockProvider<? super Path> lockProvider,
            RetryStrategy retryStrategy); @Override Path resolve(Path file); }
HadoopFileCacheRepository implements FileCacheRepository { @Override public Path resolve(Path file) throws IOException, InterruptedException { FileSystem fs = file.getFileSystem(configuration); Path qualified = fs.makeQualified(file); return doResolve(qualified); } HadoopFileCacheRepository(
            Configuration configuration,
            Path repository,
            LockProvider<? super Path> lockProvider,
            RetryStrategy retryStrategy); @Override Path resolve(Path file);  }