@Override
  public Optional<ComputationConfig> fetchConfig(String computationId) {
    Preconditions.checkArgument(
        !computationId.isEmpty(),
        "computationId is empty. Cannot fetch computation config without a computationId.");
    return fetchConfigWithRetry(
            () -> dataflowServiceClient.getStreamingConfigWorkItem(computationId))
        .flatMap(StreamingEngineComputationConfigFetcher::createComputationConfig);
  }
@Test
  public void testGetComputationConfig_noComputationPresent() throws IOException {
    Set<StreamingEnginePipelineConfig> receivedPipelineConfig = new HashSet<>();
    streamingEngineConfigFetcher =
        createConfigFetcher(/* waitForInitialConfig= */ false, 0, receivedPipelineConfig::add);
    when(mockDataflowServiceClient.getStreamingConfigWorkItem(anyString()))
        .thenReturn(Optional.empty());
    Optional<ComputationConfig> pipelineConfig =
        streamingEngineConfigFetcher.fetchConfig("someComputationId");
    assertFalse(pipelineConfig.isPresent());
    assertThat(receivedPipelineConfig).isEmpty();
  }