@VisibleForTesting
  StreamingEngineConnectionState getCurrentConnections() {
    return connections.get();
  }
@Test
  public void testStreamsStartCorrectly() throws InterruptedException {
    long items = 10L;
    long bytes = 10L;
    int numBudgetDistributionsExpected = 1;

    TestGetWorkBudgetDistributor getWorkBudgetDistributor =
        spy(new TestGetWorkBudgetDistributor(numBudgetDistributionsExpected));

    fanOutStreamingEngineWorkProvider =
        newStreamingEngineClient(
            GetWorkBudget.builder().setItems(items).setBytes(bytes).build(),
            getWorkBudgetDistributor,
            noOpProcessWorkItemFn());

    String workerToken = "workerToken1";
    String workerToken2 = "workerToken2";

    WorkerMetadataResponse firstWorkerMetadata =
        WorkerMetadataResponse.newBuilder()
            .setMetadataVersion(1)
            .addWorkEndpoints(metadataResponseEndpoint(workerToken))
            .addWorkEndpoints(metadataResponseEndpoint(workerToken2))
            .putAllGlobalDataEndpoints(DEFAULT)
            .build();

    getWorkerMetadataReady.await();
    fakeGetWorkerMetadataStub.injectWorkerMetadata(firstWorkerMetadata);
    waitForWorkerMetadataToBeConsumed(getWorkBudgetDistributor);

    StreamingEngineConnectionState currentConnections =
        fanOutStreamingEngineWorkProvider.getCurrentConnections();

    assertEquals(2, currentConnections.windmillConnections().size());
    assertEquals(2, currentConnections.windmillStreams().size());
    Set<String> workerTokens =
        currentConnections.windmillConnections().values().stream()
            .map(WindmillConnection::backendWorkerToken)
            .collect(Collectors.toSet());

    assertTrue(workerTokens.contains(workerToken));
    assertTrue(workerTokens.contains(workerToken2));

    verify(getWorkBudgetDistributor, atLeast(1))
        .distributeBudget(
            any(), eq(GetWorkBudget.builder().setItems(items).setBytes(bytes).build()));

    verify(streamFactory, times(2))
        .createDirectGetWorkStream(
            any(),
            eq(getWorkRequest(0, 0)),
            any(),
            any(),
            any(),
            any(),
            eq(noOpProcessWorkItemFn()));

    verify(streamFactory, times(2)).createGetDataStream(any(), any());
    verify(streamFactory, times(2)).createCommitWorkStream(any(), any());
  }