public synchronized ConnectionProfile createBQDestinationConnectionProfile(
      String connectionProfileId) {

    LOG.info(
        "Creating BQ Destination Connection Profile {} in project {}.",
        connectionProfileId,
        projectId);

    try {
      ConnectionProfile.Builder connectionProfileBuilder =
          ConnectionProfile.newBuilder()
              .setDisplayName(connectionProfileId)
              .setStaticServiceIpConnectivity(StaticServiceIpConnectivity.getDefaultInstance())
              .setBigqueryProfile(BigQueryProfile.newBuilder());

      CreateConnectionProfileRequest request =
          CreateConnectionProfileRequest.newBuilder()
              .setParent(LocationName.of(projectId, location).toString())
              .setConnectionProfile(connectionProfileBuilder)
              .setConnectionProfileId(connectionProfileId)
              .build();

      ConnectionProfile reference = datastreamClient.createConnectionProfileAsync(request).get();
      createdConnectionProfileIds.add(connectionProfileId);

      LOG.info(
          "Successfully created BQ Destination Connection Profile {} in project {}.",
          connectionProfileId,
          projectId);
      return reference;
    } catch (ExecutionException | InterruptedException e) {
      throw new DatastreamResourceManagerException(
          "Failed to create BQ destination connection profile. ", e);
    }
  }
@Test
  public void testCreateBQDestinationConnectionShouldCreateSuccessfully()
      throws ExecutionException, InterruptedException {
    ConnectionProfile connectionProfile = ConnectionProfile.getDefaultInstance();
    when(datastreamClient
            .createConnectionProfileAsync(any(CreateConnectionProfileRequest.class))
            .get())
        .thenReturn(connectionProfile);
    assertThat(testManager.createBQDestinationConnectionProfile(CONNECTION_PROFILE_ID))
        .isEqualTo(connectionProfile);
  }