@SuppressWarnings("FutureReturnValueIgnored")
  public void start() {
    running.set(true);
    configFetcher.start();
    memoryMonitor.start();
    streamingWorkerHarness.start();
    sampler.start();
    workerStatusReporter.start();
    activeWorkRefresher.start();
  }
@Test
  public void testOutputKeyTooLargeException() throws Exception {
    KvCoder<String, String> kvCoder = KvCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of());

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

    server.setExpectedExceptionCount(1);

    StreamingDataflowWorker worker =
        makeWorker(
            defaultWorkerParams()
                .setInstructions(instructions)
                .setOperationalLimits(
                    OperationalLimits.builder()
                        .setMaxOutputKeyBytes(15)
                        .setThrowExceptionOnLargeOutput(true)
                        .build())
                .build());
    worker.start();

    // This large key will cause the ExceptionCatchingFn to throw an exception, which will then
    // cause it to output a smaller key.
    String bigKey = "some_much_too_large_output_key";
    server.whenGetWorkCalled().thenReturn(makeInput(1, 0, bigKey, DEFAULT_SHARDING_KEY));
    server.waitForEmptyWorkQueue();

    Map<Long, Windmill.WorkItemCommitRequest> result = server.waitForAndGetCommits(1);
    assertEquals(1, result.size());
    assertEquals(
        makeExpectedOutput(1, 0, bigKey, DEFAULT_SHARDING_KEY, "smaller_key").build(),
        removeDynamicFields(result.get(1L)));
  }