@SuppressWarnings("FutureReturnValueIgnored")
  public void start() {
    running.set(true);
    configFetcher.start();
    memoryMonitor.start();
    streamingWorkerHarness.start();
    sampler.start();
    workerStatusReporter.start();
    activeWorkRefresher.start();
  }
@Test
  public void testAssignWindows() throws Exception {
    Duration gapDuration = Duration.standardSeconds(1);
    CloudObject spec = CloudObject.forClassName("AssignWindowsDoFn");
    SdkComponents sdkComponents = SdkComponents.create();
    sdkComponents.registerEnvironment(Environments.JAVA_SDK_HARNESS_ENVIRONMENT);
    addString(
        spec,
        PropertyNames.SERIALIZED_FN,
        StringUtils.byteArrayToJsonString(
            WindowingStrategyTranslation.toMessageProto(
                    WindowingStrategy.of(FixedWindows.of(gapDuration)), sdkComponents)
                .toByteArray()));

    ParallelInstruction addWindowsInstruction =
        new ParallelInstruction()
            .setSystemName("AssignWindows")
            .setName("AssignWindows")
            .setOriginalName("AssignWindowsOriginal")
            .setParDo(
                new ParDoInstruction()
                    .setInput(new InstructionInput().setProducerInstructionIndex(0).setOutputNum(0))
                    .setNumOutputs(1)
                    .setUserFn(spec))
            .setOutputs(
                Collections.singletonList(
                    new InstructionOutput()
                        .setOriginalName(DEFAULT_OUTPUT_ORIGINAL_NAME)
                        .setSystemName(DEFAULT_OUTPUT_SYSTEM_NAME)
                        .setName("output")
                        .setCodec(
                            CloudObjects.asCloudObject(
                                WindowedValue.getFullCoder(
                                    StringUtf8Coder.of(), IntervalWindow.getCoder()),
                                /* sdkComponents= */ null))));

    List<ParallelInstruction> instructions =
        Arrays.asList(
            makeSourceInstruction(StringUtf8Coder.of()),
            addWindowsInstruction,
            makeSinkInstruction(StringUtf8Coder.of(), 1));

    int timestamp1 = 0;
    int timestamp2 = 1000000;

    server
        .whenGetWorkCalled()
        .thenReturn(makeInput(timestamp1, timestamp1))
        .thenReturn(makeInput(timestamp2, timestamp2));

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

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

    assertThat(
        removeDynamicFields(result.get((long) timestamp1)),
        equalTo(
            setMessagesMetadata(
                    PaneInfo.NO_FIRING,
                    intervalWindowBytes(WINDOW_AT_ZERO),
                    makeExpectedOutput(timestamp1, timestamp1))
                .build()));

    assertThat(
        removeDynamicFields(result.get((long) timestamp2)),
        equalTo(
            setMessagesMetadata(
                    PaneInfo.NO_FIRING,
                    intervalWindowBytes(WINDOW_AT_ONE_SECOND),
                    makeExpectedOutput(timestamp2, timestamp2))
                .build()));
  }