public synchronized Stream updateStreamState(String streamId, Stream.State state) {

    LOG.info("Updating {}'s state to {} in project {}.", streamId, state.name(), projectId);

    try {
      Stream.Builder streamBuilder =
          Stream.newBuilder()
              .setName(StreamName.format(projectId, location, streamId))
              .setState(state);

      FieldMask.Builder fieldMaskBuilder = FieldMask.newBuilder().addPaths(FIELD_STATE);

      UpdateStreamRequest request =
          UpdateStreamRequest.newBuilder()
              .setStream(streamBuilder)
              .setUpdateMask(fieldMaskBuilder)
              .build();

      Stream reference = datastreamClient.updateStreamAsync(request).get();

      LOG.info(
          "Successfully updated {}'s state to {} in project {}.",
          streamId,
          state.name(),
          projectId);
      return reference;
    } catch (InterruptedException | ExecutionException e) {
      throw new DatastreamResourceManagerException("Failed to update stream. ", e);
    }
  }
@Test
  public void testUpdateStreamStateShouldCreateSuccessfully()
      throws ExecutionException, InterruptedException {
    Stream stream = Stream.getDefaultInstance();
    when(datastreamClient.updateStreamAsync(any(UpdateStreamRequest.class)).get())
        .thenReturn(stream);
    assertThat(testManager.updateStreamState(STREAM_ID, State.RUNNING)).isEqualTo(stream);
  }