public synchronized ConnectionProfile createGCSDestinationConnectionProfile(
      String connectionProfileId, String gcsBucketName, String gcsRootPath) {
    checkArgument(
        !Strings.isNullOrEmpty(connectionProfileId),
        "connectionProfileId can not be null or empty");
    checkArgument(!Strings.isNullOrEmpty(gcsBucketName), "gcsBucketName can not be null or empty");
    checkArgument(gcsRootPath != null, "gcsRootPath can not be null");
    checkArgument(
        gcsRootPath.isEmpty() || gcsRootPath.charAt(0) == '/',
        "gcsRootPath must either be an empty string or start with a '/'");

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

    try {
      ConnectionProfile.Builder connectionProfileBuilder =
          ConnectionProfile.newBuilder()
              .setDisplayName(connectionProfileId)
              .setStaticServiceIpConnectivity(StaticServiceIpConnectivity.getDefaultInstance())
              .setGcsProfile(
                  GcsProfile.newBuilder().setBucket(gcsBucketName).setRootPath(gcsRootPath));

      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 GCS Destination Connection Profile {} in project {}.",
          connectionProfileId,
          projectId);

      return reference;
    } catch (ExecutionException | InterruptedException e) {
      throw new DatastreamResourceManagerException(
          "Failed to create GCS source connection profile. ", e);
    }
  }
@Test
  public void testCreateGCSDestinationConnectionProfileWithInvalidGCSRootPathShouldFail() {
    IllegalArgumentException exception =
        assertThrows(
            IllegalArgumentException.class,
            () ->
                testManager.createGCSDestinationConnectionProfile(
                    CONNECTION_PROFILE_ID, BUCKET, "invalid"));
    assertThat(exception)
        .hasMessageThat()
        .contains("gcsRootPath must either be an empty string or start with a '/'");
  }