@SuppressWarnings("ReturnValueIgnored")
  void startStreams() {
    getWorkStream.get();
    getDataStream.get();
    commitWorkStream.get();
    workCommitter.get().start();
    // *stream.get() is all memoized in a threadsafe manner.
    started.set(true);
  }
@Test
  public void testStartStream_startsAllStreams() {
    long itemBudget = 1L;
    long byteBudget = 1L;

    WindmillStreamSender windmillStreamSender =
        newWindmillStreamSender(
            GetWorkBudget.builder().setBytes(byteBudget).setItems(itemBudget).build());

    windmillStreamSender.startStreams();

    verify(streamFactory)
        .createDirectGetWorkStream(
            eq(connection),
            eq(
                GET_WORK_REQUEST
                    .toBuilder()
                    .setMaxItems(itemBudget)
                    .setMaxBytes(byteBudget)
                    .build()),
            any(ThrottleTimer.class),
            any(),
            any(),
            any(),
            eq(workItemScheduler));

    verify(streamFactory).createGetDataStream(eq(connection.stub()), any(ThrottleTimer.class));
    verify(streamFactory).createCommitWorkStream(eq(connection.stub()), any(ThrottleTimer.class));
  }