@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_fetchConfigFromDataflowError() throws IOException {
    streamingEngineConfigFetcher =
        createConfigFetcher(/* waitForInitialConfig= */ false, 0, ignored -> {});
    RuntimeException e = new RuntimeException("something bad happened.");
    when(mockDataflowServiceClient.getStreamingConfigWorkItem(anyString())).thenThrow(e);
    Throwable fetchConfigError =
        assertThrows(
            RuntimeException.class,
            () -> streamingEngineConfigFetcher.fetchConfig("someComputationId"));
    assertThat(fetchConfigError).isSameInstanceAs(e);
  }