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

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

    for (int i = 0; i < 2; i++) {
      server
          .whenGetWorkCalled()
          .thenReturn(
              makeInput(
                  i, TimeUnit.MILLISECONDS.toMicros(i), keyStringForIndex(i), DEFAULT_SHARDING_KEY))
          .thenReturn(
              makeInput(
                  i + 1000,
                  TimeUnit.MILLISECONDS.toMicros(i),
                  keyStringForIndex(i),
                  DEFAULT_SHARDING_KEY + i));
    }

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

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

    for (int i = 0; i < 2; i++) {
      assertTrue(result.containsKey((long) i));
      assertEquals(
          makeExpectedOutput(
                  i,
                  TimeUnit.MILLISECONDS.toMicros(i),
                  keyStringForIndex(i),
                  DEFAULT_SHARDING_KEY,
                  keyStringForIndex(i) + "_data" + i)
              .build(),
          removeDynamicFields(result.get((long) i)));
      assertTrue(result.containsKey((long) i + 1000));
      assertEquals(
          makeExpectedOutput(
                  i + 1000,
                  TimeUnit.MILLISECONDS.toMicros(i),
                  keyStringForIndex(i),
                  DEFAULT_SHARDING_KEY + i,
                  keyStringForIndex(i) + "_data" + (i + 1000))
              .build(),
          removeDynamicFields(result.get((long) i + 1000)));
    }
  }