public static <ID, T> TaskDispatcher<ID, T> createBatchingTaskDispatcher(String id,
                                                                             int maxBufferSize,
                                                                             int workloadSize,
                                                                             int workerCount,
                                                                             long maxBatchingDelay,
                                                                             long congestionRetryDelayMs,
                                                                             long networkFailureRetryMs,
                                                                             TaskProcessor<T> taskProcessor) {
        final AcceptorExecutor<ID, T> acceptorExecutor = new AcceptorExecutor<>(
                id, maxBufferSize, workloadSize, maxBatchingDelay, congestionRetryDelayMs, networkFailureRetryMs
        );
        final TaskExecutors<ID, T> taskExecutor = TaskExecutors.batchExecutors(id, workerCount, taskProcessor, acceptorExecutor);
        return new TaskDispatcher<ID, T>() {
            @Override
            public void process(ID id, T task, long expiryTime) {
                acceptorExecutor.process(id, task, expiryTime);
            }

            @Override
            public void shutdown() {
                acceptorExecutor.shutdown();
                taskExecutor.shutdown();
            }
        };
    }
@Test
    public void testTasksAreDistributedAcrossAllWorkerThreads() throws Exception {
        int threadCount = 3;
        CountingTaskProcessor countingProcessor = new CountingTaskProcessor();

        TaskDispatcher<Integer, Boolean> dispatcher = TaskDispatchers.createBatchingTaskDispatcher(
                "TEST",
                MAX_BUFFER_SIZE,
                WORK_LOAD_SIZE,
                threadCount,
                MAX_BATCHING_DELAY_MS,
                SERVER_UNAVAILABLE_SLEEP_TIME_MS,
                RETRY_SLEEP_TIME_MS,
                countingProcessor
        );

        try {
            int loops = 1000;
            while (true) {
                countingProcessor.resetTo(loops);
                for (int i = 0; i < loops; i++) {
                    dispatcher.process(i, true, System.currentTimeMillis() + 60 * 1000);
                }
                countingProcessor.awaitCompletion();
                int minHitPerThread = (int) (loops / threadCount * 0.9);
                if (countingProcessor.lowestHit() < minHitPerThread) {
                    loops *= 2;
                } else {
                    break;
                }
                if (loops > MAX_BUFFER_SIZE) {
                    fail("Uneven load distribution");
                }
            }
        } finally {
            dispatcher.shutdown();
        }
    }