TaskRunnerConfigurer { public synchronized void init() { MetricsContainer.incrementInitializationCount(this.getClass().getCanonicalName()); this.taskPollExecutor = new TaskPollExecutor(eurekaClient, taskClient, threadCount, updateRetryCount, taskToDomain, workerNamePrefix); this.scheduledExecutorService = Executors.newScheduledThreadPool(workers.size()); workers.forEach( worker -> scheduledExecutorService.scheduleWithFixedDelay(() -> taskPollExecutor.pollAndExecute(worker), worker.getPollingInterval(), worker.getPollingInterval(), TimeUnit.MILLISECONDS)); } private  TaskRunnerConfigurer(Builder builder); int getThreadCount(); int getSleepWhenRetry(); int getUpdateRetryCount(); String getWorkerNamePrefix(); synchronized void init(); void shutdown();  }
@Test public void testMultipleWorkersExecution() { String task1Name = "task1"; Worker worker1 = mock(Worker.class); when(worker1.getPollingInterval()).thenReturn(3000); when(worker1.getTaskDefName()).thenReturn(task1Name); when(worker1.getIdentity()).thenReturn("worker1"); when(worker1.execute(any())).thenAnswer(invocation -> { Uninterruptibles.sleepUninterruptibly(2, TimeUnit.SECONDS); TaskResult taskResult = new TaskResult(); taskResult.setStatus(COMPLETED); return taskResult; }); String task2Name = "task2"; Worker worker2 = mock(Worker.class); when(worker2.getPollingInterval()).thenReturn(3000); when(worker2.getTaskDefName()).thenReturn(task2Name); when(worker2.getIdentity()).thenReturn("worker2"); when(worker2.execute(any())).thenAnswer(invocation -> { Uninterruptibles.sleepUninterruptibly(2, TimeUnit.SECONDS); TaskResult taskResult = new TaskResult(); taskResult.setStatus(COMPLETED); return taskResult; }); Task task1 = testTask(task1Name); Task task2 = testTask(task2Name); TaskClient taskClient = Mockito.mock(TaskClient.class); TaskRunnerConfigurer configurer = new TaskRunnerConfigurer.Builder(taskClient, Arrays.asList(worker1, worker2)) .withThreadCount(2) .withSleepWhenRetry(100000) .withUpdateRetryCount(1) .withWorkerNamePrefix("test-worker-") .build(); when(taskClient.pollTask(any(), any(), any())).thenAnswer(invocation -> { Object[] args = invocation.getArguments(); String taskName = args[0].toString(); if (taskName.equals(task1Name)) { return task1; } else if (taskName.equals(task2Name)) { return task2; } else { return null; } }); when(taskClient.ack(any(), any())).thenReturn(true); AtomicInteger task1Counter = new AtomicInteger(0); AtomicInteger task2Counter = new AtomicInteger(0); CountDownLatch latch = new CountDownLatch(2); doAnswer(invocation -> { Object[] args = invocation.getArguments(); TaskResult result = (TaskResult) args[0]; assertEquals(COMPLETED, result.getStatus()); if (result.getWorkerId().equals("worker1")) { task1Counter.incrementAndGet(); } else if (result.getWorkerId().equals("worker2")) { task2Counter.incrementAndGet(); } latch.countDown(); return null; } ).when(taskClient).updateTask(any()); configurer.init(); Uninterruptibles.awaitUninterruptibly(latch); assertEquals(1, task1Counter.get()); assertEquals(1, task2Counter.get()); }