@Override
  public String addResource(String key, String fileName) {
    String interned = intern(key);
    synchronized (interned) {
      SharedCacheResource resource = cachedResources.get(interned);
      if (resource == null) {
        resource = new SharedCacheResource(fileName);
        cachedResources.put(interned, resource);
      }
      return resource.getFileName();
    }
  }
@Test
  void testAddResourceConcurrency() throws Exception {
    startEmptyStore();
    final String key = "key1";
    int count = 5;
    ExecutorService exec = HadoopExecutors.newFixedThreadPool(count);
    List<Future<String>> futures = new ArrayList<Future<String>>(count);
    final CountDownLatch start = new CountDownLatch(1);
    for (int i = 0; i < count; i++) {
      final String fileName = "foo-" + i + ".jar";
      Callable<String> task = new Callable<String>() {
        public String call() throws Exception {
          start.await();
          String result = store.addResource(key, fileName);
          System.out.println("fileName: " + fileName + ", result: " + result);
          return result;
        }
      };
      futures.add(exec.submit(task));
    }
    // start them all at the same time
    start.countDown();
    // check the result; they should all agree with the value
    Set<String> results = new HashSet<String>();
    for (Future<String> future : futures) {
      results.add(future.get());
    }
    assertSame(1, results.size());
    exec.shutdown();
  }