@SuppressWarnings("FutureReturnValueIgnored")
  public void start() {
    running.set(true);
    configFetcher.start();
    memoryMonitor.start();
    streamingWorkerHarness.start();
    sampler.start();
    workerStatusReporter.start();
    activeWorkRefresher.start();
  }
@Test
  public void testLimitOnOutputBundleSize() throws Exception {
    // This verifies that ReadOperation, StreamingModeExecutionContext, and windmill sinks
    // coordinate to limit size of an output bundle.
    List<Integer> finalizeTracker = Lists.newArrayList();
    TestCountingSource.setFinalizeTracker(finalizeTracker);

    final int numMessagesInCustomSourceShard = 100000; // 100K input messages.
    final int inflatedSizePerMessage = 10000; // x10k => 1GB total output size.

    StreamingDataflowWorker worker =
        makeWorker(
            defaultWorkerParams()
                .setInstructions(
                    makeUnboundedSourcePipeline(
                        numMessagesInCustomSourceShard, new InflateDoFn(inflatedSizePerMessage)))
                .build());
    worker.start();

    // Test new key.
    server
        .whenGetWorkCalled()
        .thenReturn(
            buildInput(
                "work {"
                    + "  computation_id: \"computation\""
                    + "  input_data_watermark: 0"
                    + "  work {"
                    + "    key: \"0000000000000001\""
                    + "    sharding_key: 1"
                    + "    work_token: 1"
                    + "    cache_token: 1"
                    + "  }"
                    + "}",
                null));

    // Matcher to ensure that commit size is within 10% of max bundle size.
    Matcher<Integer> isWithinBundleSizeLimits =
        both(greaterThan(StreamingDataflowWorker.MAX_SINK_BYTES * 9 / 10))
            .and(lessThan(StreamingDataflowWorker.MAX_SINK_BYTES * 11 / 10));

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

    assertThat(commit.getSerializedSize(), isWithinBundleSizeLimits);

    // Try another bundle
    server
        .whenGetWorkCalled()
        .thenReturn(
            buildInput(
                "work {"
                    + "  computation_id: \"computation\""
                    + "  input_data_watermark: 0"
                    + "  work {"
                    + "    key: \"0000000000000001\""
                    + "    sharding_key: 1"
                    + "    work_token: 2"
                    + "    cache_token: 1"
                    + "  }"
                    + "}",
                null));

    result = server.waitForAndGetCommits(1);
    commit = result.get(2L);

    assertThat(commit.getSerializedSize(), isWithinBundleSizeLimits);
  }