@SuppressWarnings("FutureReturnValueIgnored")
  public void start() {
    running.set(true);
    configFetcher.start();
    memoryMonitor.start();
    streamingWorkerHarness.start();
    sampler.start();
    workerStatusReporter.start();
    activeWorkRefresher.start();
  }
@Test(timeout = 10000)
  public void testNumberOfWorkerHarnessThreadsIsHonored() throws Exception {
    int expectedNumberOfThreads = 5;
    List<ParallelInstruction> instructions =
        Arrays.asList(
            makeSourceInstruction(StringUtf8Coder.of()),
            makeDoFnInstruction(blockingFn, 0, StringUtf8Coder.of()),
            makeSinkInstruction(StringUtf8Coder.of(), 0));

    StreamingDataflowWorker worker =
        makeWorker(
            defaultWorkerParams("--numberOfWorkerHarnessThreads=" + expectedNumberOfThreads)
                .setInstructions(instructions)
                .publishCounters()
                .build());
    worker.start();

    for (int i = 0; i < expectedNumberOfThreads * 2; ++i) {
      server.whenGetWorkCalled().thenReturn(makeInput(i, TimeUnit.MILLISECONDS.toMicros(i)));
    }

    // This will fail to complete if the number of threads is less than the amount of work.
    // Forcing this test to timeout.
    BlockingFn.counter.acquire(expectedNumberOfThreads);

    // Attempt to acquire an additional permit, if we were able to then that means
    // too many items were being processed concurrently.
    if (BlockingFn.counter.tryAcquire(500, TimeUnit.MILLISECONDS)) {
      fail(
          "Expected number of threads "
              + expectedNumberOfThreads
              + " does not match actual "
              + "number of work items processed concurrently "
              + BlockingFn.callCounter.get()
              + ".");
    }

    BlockingFn.blocker.countDown();
  }