@Override
  public void start() {
    fetchInitialPipelineGlobalConfig();
    schedulePeriodicGlobalConfigRequests();
  }
@Test
  public void testStart_requiresInitialConfig() throws IOException, InterruptedException {
    WorkItem initialConfig =
        new WorkItem()
            .setJobId("job")
            .setStreamingConfigTask(new StreamingConfigTask().setMaxWorkItemCommitBytes(10L));
    CountDownLatch waitForInitialConfig = new CountDownLatch(1);
    Set<StreamingEnginePipelineConfig> receivedPipelineConfig = new HashSet<>();
    when(mockDataflowServiceClient.getGlobalStreamingConfigWorkItem())
        .thenReturn(Optional.of(initialConfig));
    streamingEngineConfigFetcher =
        createConfigFetcher(
            /* waitForInitialConfig= */ true,
            0,
            config -> {
              try {
                receivedPipelineConfig.add(config);
                waitForInitialConfig.await();
              } catch (InterruptedException e) {
                throw new RuntimeException(e);
              }
            });
    Thread asyncStartConfigLoader = new Thread(streamingEngineConfigFetcher::start);
    asyncStartConfigLoader.start();
    waitForInitialConfig.countDown();
    asyncStartConfigLoader.join();
    assertThat(receivedPipelineConfig)
        .containsExactly(
            StreamingEnginePipelineConfig.builder()
                .setMaxWorkItemCommitBytes(
                    initialConfig.getStreamingConfigTask().getMaxWorkItemCommitBytes())
                .build());
  }