### Question:
ResourceLimitCache implements Cache<K, V> { public ResourceLimitCache(java.util.concurrent.ConcurrentMap<K, V> map, BiFunction<K, V, int[]> permitsSupplier, int... limits) { Objects.requireNonNull(map); Objects.requireNonNull(permitsSupplier); Objects.requireNonNull(limits); this.map = map; this.group = new ResourceSemaphore.Group(limits); this.permitsSupplier = permitsSupplier; } ResourceLimitCache(java.util.concurrent.ConcurrentMap<K, V> map,
      BiFunction<K, V, int[]> permitsSupplier, int... limits); @Override V get(K key); @Override V put(K key, V value); @Override V remove(K key); @Override void removeIf(Predicate<K> predicate); @Override void clear();  }### Answer:
@Test public void testResourceLimitCache() throws InterruptedException, TimeoutException { Cache<Integer, String> resourceCache = new ResourceLimitCache<>(new ConcurrentHashMap<>(), (k, v) -> new int[] {k}, 10); resourceCache.put(6, "a"); resourceCache.put(4, "a"); resourceCache.put(4, "a"); CompletableFuture future = CompletableFuture.supplyAsync(() -> { try { return resourceCache.put(1, "a"); } catch (InterruptedException e) { e.printStackTrace(); } return null; }); Assert.assertTrue(!future.isDone()); Thread.sleep(100); Assert.assertTrue(!future.isDone()); resourceCache.remove(4); GenericTestUtils.waitFor(future::isDone, 100, 1000); Assert.assertTrue(future.isDone() && !future.isCompletedExceptionally()); Assert.assertNotNull(resourceCache.get(1)); ExecutorService pool = Executors.newCachedThreadPool(); future = CompletableFuture.supplyAsync(() -> { try { return resourceCache.put(4, "a"); } catch (InterruptedException e) { return null; } }, pool); Assert.assertTrue(!future.isDone()); Thread.sleep(100); Assert.assertTrue(!future.isDone()); pool.shutdownNow(); future.cancel(true); resourceCache.remove(1); Assert.assertNull(resourceCache.get(4)); }