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 testBatchingDispatcher() throws Exception {
        dispatcher = TaskDispatchers.createBatchingTaskDispatcher(
                "TEST",
                MAX_BUFFER_SIZE,
                WORK_LOAD_SIZE,
                1,
                MAX_BATCHING_DELAY_MS,
                SERVER_UNAVAILABLE_SLEEP_TIME_MS,
                RETRY_SLEEP_TIME_MS,
                processor
        );

        dispatcher.process(1, ProcessingResult.Success, System.currentTimeMillis() + 60 * 1000);
        dispatcher.process(2, ProcessingResult.Success, System.currentTimeMillis() + 60 * 1000);

        processor.expectSuccesses(2);
    }