### Question:
ClusterAwareExecutor { public synchronized void submit( @Nonnull final String clusterName, @Nonnull final String operation, @Nonnull final Runnable runnable) { if (null == taskQueue.putIfAbsent(new ClusterOperation(clusterName, operation), runnable)) { dispatch(); } } ClusterAwareExecutor(@Nonnull final ExecutorService executorService, @Nonnull final NamedSemaphores namedSemaphores); synchronized void submit(
            @Nonnull final String clusterName,
            @Nonnull final String operation,
            @Nonnull final Runnable runnable);  }### Answer:
@Test void deduplication() throws InterruptedException { final ClusterAwareExecutor executor = new ClusterAwareExecutor(EXECUTOR_SERVICE, NAMED_LOCKS); final AtomicLong executionCount = new AtomicLong(); final CountDownLatch startLatch = new CountDownLatch(1); final CountDownLatch completionLatch = new CountDownLatch(2); for (int index = 0; index < 10; index++) { executor.submit("cluster", "operation", () -> { try { startLatch.await(10, TimeUnit.SECONDS); } catch (final InterruptedException ignored) {} executionCount.incrementAndGet(); completionLatch.countDown(); }); } startLatch.countDown(); completionLatch.await(10, TimeUnit.SECONDS); assertEquals(2, executionCount.get()); }
@Test void oneActiveTaskPerCluster() throws InterruptedException { final ClusterAwareExecutor executor = new ClusterAwareExecutor(EXECUTOR_SERVICE, NAMED_LOCKS); final CountDownLatch cluster1StartLatch = new CountDownLatch(1); final CountDownLatch cluster1CompletionLatch = new CountDownLatch(2); final CountDownLatch cluster2CompletionLatch = new CountDownLatch(1); for (int index = 0; index < 2; index++) { executor.submit("cluster1", String.format("operation-%d", index), () -> { try { cluster1StartLatch.await(10, TimeUnit.SECONDS); } catch (final InterruptedException ignored) {} cluster1CompletionLatch.countDown(); }); } executor.submit("cluster2", "operation", cluster2CompletionLatch::countDown); cluster2CompletionLatch.await(10, TimeUnit.SECONDS); assertEquals(2, cluster1CompletionLatch.getCount()); cluster1StartLatch.countDown(); cluster1CompletionLatch.await(10, TimeUnit.SECONDS); }