@SuppressWarnings("FutureReturnValueIgnored")
  public void start() {
    running.set(true);
    configFetcher.start();
    memoryMonitor.start();
    streamingWorkerHarness.start();
    sampler.start();
    workerStatusReporter.start();
    activeWorkRefresher.start();
  }
@Test
  public void testKeyTokenInvalidException() throws Exception {
    if (streamingEngine) {
      // TODO: This test needs to be adapted to work with streamingEngine=true.
      return;
    }
    KvCoder<String, String> kvCoder = KvCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of());

    List<ParallelInstruction> instructions =
        Arrays.asList(
            makeSourceInstruction(kvCoder),
            makeDoFnInstruction(new KeyTokenInvalidFn(), 0, kvCoder),
            makeSinkInstruction(kvCoder, 1));

    server
        .whenGetWorkCalled()
        .thenReturn(makeInput(0, 0, DEFAULT_KEY_STRING, DEFAULT_SHARDING_KEY));

    StreamingDataflowWorker worker =
        makeWorker(defaultWorkerParams().setInstructions(instructions).publishCounters().build());
    worker.start();

    server.waitForEmptyWorkQueue();

    server
        .whenGetWorkCalled()
        .thenReturn(makeInput(1, 0, DEFAULT_KEY_STRING, DEFAULT_SHARDING_KEY));

    Map<Long, Windmill.WorkItemCommitRequest> result = server.waitForAndGetCommits(1);

    assertEquals(
        makeExpectedOutput(1, 0, DEFAULT_KEY_STRING, DEFAULT_SHARDING_KEY, DEFAULT_KEY_STRING)
            .build(),
        removeDynamicFields(result.get(1L)));
    assertEquals(1, result.size());
  }