@SuppressWarnings("FutureReturnValueIgnored")
  public void start() {
    running.set(true);
    configFetcher.start();
    memoryMonitor.start();
    streamingWorkerHarness.start();
    sampler.start();
    workerStatusReporter.start();
    activeWorkRefresher.start();
  }
@Test
  public void testOutputValueTooLargeException() 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()
                        .setMaxOutputValueBytes(15)
                        .setThrowExceptionOnLargeOutput(true)
                        .build())
                .build());
    worker.start();

    // The first time processing will have value "data1_a_bunch_more_data_output", which is above
    // the limit. After throwing the exception, the output should be just "data1", which is small
    // enough.
    server.whenGetWorkCalled().thenReturn(makeInput(1, 0, "key", DEFAULT_SHARDING_KEY));
    server.waitForEmptyWorkQueue();

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