@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);
    // We start the appropriate number of threads.
    for (int i = 0; i < numThreads; ++i) {
      tasks.add(new WorkerThread(worker, sleeper));
    }

    LOG.debug("Waiting for {} worker threads", numThreads);
    // We wait forever unless there is a big problem.
    executor.invokeAll(tasks);
    LOG.error("All threads died.");
  }
@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.");
        });
    // Verify that the number of requested worker harness threads is honored.
    assertEquals(expectedNumberOfThreads, fakeWorker.getNumberOfCallsToGetAndPerformWork());
  }