@Override
  public void start() {
    fetchInitialPipelineGlobalConfig();
    schedulePeriodicGlobalConfigRequests();
  }
@Test
  public void testStart_startsPeriodicConfigRequests() throws IOException, InterruptedException {
    WorkItem firstConfig =
        new WorkItem()
            .setJobId("job")
            .setStreamingConfigTask(new StreamingConfigTask().setMaxWorkItemCommitBytes(10L));
    WorkItem secondConfig =
        new WorkItem()
            .setJobId("job")
            .setStreamingConfigTask(new StreamingConfigTask().setMaxWorkItemCommitBytes(15L));
    WorkItem thirdConfig =
        new WorkItem()
            .setJobId("job")
            .setStreamingConfigTask(new StreamingConfigTask().setMaxWorkItemCommitBytes(100L));
    CountDownLatch numExpectedRefreshes = new CountDownLatch(3);
    Set<StreamingEnginePipelineConfig> receivedPipelineConfig = new HashSet<>();
    when(mockDataflowServiceClient.getGlobalStreamingConfigWorkItem())
        .thenReturn(Optional.of(firstConfig))
        .thenReturn(Optional.of(secondConfig))
        // ConfigFetcher should still fetch subsequent configs on error.
        .thenThrow(new IOException("something bad happened."))
        .thenReturn(Optional.of(thirdConfig))
        // ConfigFetcher should not do anything with a config that doesn't contain a
        // StreamingConfigTask.
        .thenReturn(Optional.of(new WorkItem().setJobId("jobId")));

    streamingEngineConfigFetcher =
        createConfigFetcher(
            /* waitForInitialConfig= */ true,
            Duration.millis(100).getMillis(),
            config -> {
              receivedPipelineConfig.add(config);
              numExpectedRefreshes.countDown();
            });

    Thread asyncStartConfigLoader = new Thread(streamingEngineConfigFetcher::start);
    asyncStartConfigLoader.start();
    numExpectedRefreshes.await();
    asyncStartConfigLoader.join();
    assertThat(receivedPipelineConfig)
        .containsExactly(
            StreamingEnginePipelineConfig.builder()
                .setMaxWorkItemCommitBytes(
                    firstConfig.getStreamingConfigTask().getMaxWorkItemCommitBytes())
                .build(),
            StreamingEnginePipelineConfig.builder()
                .setMaxWorkItemCommitBytes(
                    secondConfig.getStreamingConfigTask().getMaxWorkItemCommitBytes())
                .build(),
            StreamingEnginePipelineConfig.builder()
                .setMaxWorkItemCommitBytes(
                    thirdConfig.getStreamingConfigTask().getMaxWorkItemCommitBytes())
                .build());
  }