public synchronized Stream createStream(
      String streamId, SourceConfig sourceConfig, DestinationConfig destinationConfig) {

    LOG.info("Creating Stream {} in project {}.", streamId, projectId);

    try {
      CreateStreamRequest request =
          CreateStreamRequest.newBuilder()
              .setParent(LocationName.format(projectId, location))
              .setStreamId(streamId)
              .setStream(
                  Stream.newBuilder()
                      .setSourceConfig(sourceConfig)
                      .setDisplayName(streamId)
                      .setDestinationConfig(destinationConfig)
                      .setBackfillAll(Stream.BackfillAllStrategy.getDefaultInstance())
                      .build())
              .build();

      Stream reference = datastreamClient.createStreamAsync(request).get();
      createdStreamIds.add(streamId);

      LOG.info("Successfully created Stream {} in project {}.", streamId, projectId);
      return reference;
    } catch (ExecutionException | InterruptedException e) {
      throw new DatastreamResourceManagerException("Failed to create stream. ", e);
    }
  }
@Test
  public void testCreateStreamExecutionExceptionShouldFail()
      throws ExecutionException, InterruptedException {
    when(datastreamClient.createStreamAsync(any(CreateStreamRequest.class)).get())
        .thenThrow(ExecutionException.class);
    DatastreamResourceManagerException exception =
        assertThrows(
            DatastreamResourceManagerException.class,
            () ->
                testManager.createStream(
                    STREAM_ID,
                    SourceConfig.getDefaultInstance(),
                    DestinationConfig.getDefaultInstance()));
    assertThat(exception).hasMessageThat().contains("Failed to create stream.");
  }