class DefaultBucket {

  public static String tryCreateDefaultBucket(PipelineOptions options) {
    GcsOptions gcpOptions = options.as(GcsOptions.class);

    final String projectId = gcpOptions.getProject();
    checkArgument(!isNullOrEmpty(projectId),
                  "--project is a required option.");

    // Look up the project number, to create a default bucket with a stable
    // name with no special characters.
    long projectNumber = 0L;
    try {
      projectNumber = gcpOptions.as(CloudResourceManagerOptions.class)
          .getGcpProjectUtil().getProjectNumber(projectId);
    } catch (IOException e) {
      throw new RuntimeException("Unable to verify project with ID " + projectId, e);
    }
    String region = DEFAULT_REGION;
    if (!isNullOrEmpty(gcpOptions.getZone())) {
      region = getRegionFromZone(gcpOptions.getZone());
    }
    final String bucketName =
      "dataflow-staging-" + region + "-" + projectNumber;
    LOG.info("No staging location provided, attempting to use default bucket: {}",
             bucketName);
    Bucket bucket = new Bucket()
      .setName(bucketName)
      .setLocation(region);
    // Always try to create the bucket before checking access, so that we do not
    // race with other pipelines that may be attempting to do the same thing.
    try {
      gcpOptions.getGcsUtil().createBucket(projectId, bucket);
    } catch (FileAlreadyExistsException e) {
      LOG.debug("Bucket '{}'' already exists, verifying access.", bucketName);
    } catch (IOException e) {
      throw new RuntimeException("Unable create default bucket.", e);
    }

    // Once the bucket is expected to exist, verify that it is correctly owned
    // by the project executing the job.
    try {
      long owner = gcpOptions.getGcsUtil().bucketOwner(
        GcsPath.fromComponents(bucketName, ""));
      checkArgument(
        owner == projectNumber,
        "Bucket owner does not match the project from --project:"
        + " %s vs. %s", owner, projectNumber);
    } catch (IOException e) {
      throw new RuntimeException(
        "Unable to determine the owner of the default bucket at gs://" + bucketName, e);
    }
    return "gs://" + bucketName;
  }

}