@SuppressWarnings("FutureReturnValueIgnored")
  public void start() {
    running.set(true);
    configFetcher.start();
    memoryMonitor.start();
    streamingWorkerHarness.start();
    sampler.start();
    workerStatusReporter.start();
    activeWorkRefresher.start();
  }
@Test
  public void testUnboundedSourcesDrain() throws Exception {
    List<Integer> finalizeTracker = Lists.newArrayList();
    TestCountingSource.setFinalizeTracker(finalizeTracker);

    StreamingDataflowWorker worker =
        makeWorker(
            defaultWorkerParams()
                .setInstructions(makeUnboundedSourcePipeline())
                .publishCounters()
                .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: 2"
                    + "    cache_token: 3"
                    + "  }"
                    + "}",
                null));

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

    Windmill.WorkItemCommitRequest commit = result.get(2L);
    UnsignedLong finalizeId =
        UnsignedLong.fromLongBits(commit.getSourceStateUpdates().getFinalizeIds(0));

    assertThat(
        removeDynamicFields(commit),
        equalTo(
            setMessagesMetadata(
                    PaneInfo.NO_FIRING,
                    CoderUtils.encodeToByteArray(
                        CollectionCoder.of(GlobalWindow.Coder.INSTANCE),
                        Collections.singletonList(GlobalWindow.INSTANCE)),
                    parseCommitRequest(
                        "key: \"0000000000000001\" "
                            + "sharding_key: 1 "
                            + "work_token: 2 "
                            + "cache_token: 3 "
                            + "source_backlog_bytes: 7 "
                            + "source_bytes_processed: 18 "
                            + "output_messages {"
                            + "  destination_stream_id: \"out\""
                            + "  bundles {"
                            + "    key: \"0000000000000001\""
                            + "    messages {"
                            + "      timestamp: 0"
                            + "      data: \"0:0\""
                            + "    }"
                            + "    messages_ids: \"\""
                            + "  }"
                            + "} "
                            + "source_state_updates {"
                            + "  state: \"\000\""
                            + "  finalize_ids: "
                            + finalizeId
                            + "} "
                            + "source_watermark: 1000"))
                .build()));

    // Test drain work item.
    server
        .whenGetWorkCalled()
        .thenReturn(
            buildInput(
                "work {"
                    + "  computation_id: \"computation\""
                    + "  input_data_watermark: 0"
                    + "  work {"
                    + "    key: \"0000000000000001\""
                    + "    sharding_key: 1"
                    + "    work_token: 3"
                    + "    cache_token: 3"
                    + "    source_state {"
                    + "      only_finalize: true"
                    + "      finalize_ids: "
                    + finalizeId
                    + "    }"
                    + "  }"
                    + "}",
                null));

    result = server.waitForAndGetCommits(1);

    commit = result.get(3L);

    assertThat(
        commit,
        equalTo(
            parseCommitRequest(
                    "key: \"0000000000000001\" "
                        + "sharding_key: 1 "
                        + "work_token: 3 "
                        + "cache_token: 3 "
                        + "source_state_updates {"
                        + "  only_finalize: true"
                        + "} ")
                .build()));

    assertThat(finalizeTracker, contains(0));
  }