DataflowBatchWorkerHarness { @VisibleForTesting static void processWork( DataflowWorkerHarnessOptions pipelineOptions, final BatchDataflowWorker worker, Sleeper sleeper) throws InterruptedException { int numThreads = chooseNumberOfThreads(pipelineOptions); ExecutorService executor = pipelineOptions.getExecutorService(); final List<Callable<Boolean>> tasks = new ArrayList<>(); LOG.debug("Starting {} worker threads", numThreads); for (int i = 0; i < numThreads; ++i) { tasks.add(new WorkerThread(worker, sleeper)); } LOG.debug("Waiting for {} worker threads", numThreads); executor.invokeAll(tasks); LOG.error("All threads died."); } private  DataflowBatchWorkerHarness(DataflowWorkerHarnessOptions pipelineOptions); static void main(String[] args); static DataflowBatchWorkerHarness from(DataflowWorkerHarnessOptions pipelineOptions); void run();  }
@Test public void testNumberOfWorkerHarnessThreadsIsHonored() throws Exception { final int expectedNumberOfThreads = 5; pipelineOptions.setNumberOfWorkerHarnessThreads(expectedNumberOfThreads); FakeWorker fakeWorker = new FakeWorker(pipelineOptions, false); DataflowBatchWorkerHarness.processWork( pipelineOptions, fakeWorker, millis -> { throw new InterruptedException("Stopping the retry loop."); }); assertEquals(expectedNumberOfThreads, fakeWorker.getNumberOfCallsToGetAndPerformWork()); }