ScheduledExecutorServiceMock implements ScheduledExecutorService { @Override public <T> T invokeAny(final Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException { try { return doInvokeAny(tasks, null, null); } catch (final TimeoutException e) { throw new RuntimeException("Unexpected timeout exception", e); } } ScheduledExecutorServiceMock();  ScheduledExecutorServiceMock(final ISystemTimeProvider systemTimeProvider); List<ScheduledFutureMock<?>> getEvents(); List<ScheduledFutureMock<?>> executeEvents(); ScheduledFutureMock<?> executeNextEvent(); ScheduledFutureMock<?> offerNextEventForExecution(); @Override void execute(final Runnable runnable); @Override ScheduledFutureMock<Void> submit(final Runnable runnable); @Override ScheduledFutureMock<T> submit(final Runnable runnable, final T result); @Override ScheduledFutureMock<T> submit(final Callable<T> callable); @Override ScheduledFutureMock<Void> schedule(final Runnable runnable, final long delay, final TimeUnit unit); @Override ScheduledFutureMock<T> schedule(final Callable<T> callable, final long delay, final TimeUnit unit); @Override ScheduledFutureMock<Void> scheduleAtFixedRate(
        final Runnable runnable,
        final long initialDelay,
        final long period,
        final TimeUnit unit); @Override ScheduledFutureMock<Void> scheduleWithFixedDelay(
        final Runnable runnable,
        final long initialDelay,
        final long delay,
        final TimeUnit unit); @Override T invokeAny(final Collection<? extends Callable<T>> tasks); @Override T invokeAny(final Collection<? extends Callable<T>> tasks, final long timeout, final TimeUnit unit); @Override List<Future<T>> invokeAll(final Collection<? extends Callable<T>> tasks, final long timeout, final TimeUnit unit); @Override List<Future<T>> invokeAll(final Collection<? extends Callable<T>> tasks); @Override void shutdown(); @Override List<Runnable> shutdownNow(); @Override boolean isShutdown(); @Override boolean isTerminated(); @Override boolean awaitTermination(final long timeout, final TimeUnit unit);  }
@Test @Ignore("Has unfixed deadlock sometimes") public void testInvokeAny() throws InterruptedException, ExecutionException { final ScheduledExecutorServiceMock executor = new ScheduledExecutorServiceMock(); final int taskCount = 10; final int dispatcherCount = 5; final ArrayList<Callable<Integer>> tasks = new ArrayList<Callable<Integer>>(taskCount); for (int i = 0; i < taskCount; i++) { final Integer result = Integer.valueOf(i); tasks.add(new Callable<Integer>() { @Override public Integer call() throws Exception { if (result.intValue() > dispatcherCount / 2) { Thread.sleep(50000); } return result; } }); } final CountDownLatch dispatcherSync = new CountDownLatch(1); final ArrayList<MessageDispatcher> messageDispatchers = new ArrayList<MessageDispatcher>(); final ArrayList<Thread> dispatcherThreads = new ArrayList<Thread>(); for (int i = 0; i < dispatcherCount; i++) { final MessageDispatcher messageDispatcher = new MessageDispatcher(executor, dispatcherSync); messageDispatchers.add(messageDispatcher); final Thread dispatcherThread = new Thread(messageDispatcher, "Dispatcher thread" + i); dispatcherThreads.add(dispatcherThread); dispatcherThread.start(); } final List<ScheduledFutureMock<?>> scheduledTasks = new LinkedList<ScheduledFutureMock<?>>(); final Thread tasksAdderThread = new Thread(new Runnable() { @Override public void run() { List<ScheduledFutureMock<?>> events = executor.getEvents(); while (events.size() != taskCount) { events = executor.getEvents(); } scheduledTasks.addAll(events); dispatcherSync.countDown(); } }, "Events adder thread"); tasksAdderThread.start(); try { final Integer invocationResult = executor.invokeAny(tasks); tasksAdderThread.join(); Assert.assertEquals(taskCount, scheduledTasks.size()); for (final ScheduledFutureMock<?> future : scheduledTasks) { Assert.assertTrue(future.isDone()); } Assert.assertNotNull(invocationResult); Assert.assertTrue(invocationResult.intValue() >= 0); Assert.assertTrue(invocationResult.intValue() < taskCount); } finally { executor.shutdown(); Assert.assertTrue(executor.awaitTermination(TIMEOUT, TimeUnit.MILLISECONDS)); Assert.assertEquals(0, executor.getEvents().size()); for (final Thread dispatcherThread : dispatcherThreads) { dispatcherThread.join(TIMEOUT); Assert.assertFalse(dispatcherThread.isAlive()); } } }