Initialze storage client with project from pipeline option.

CHANGES.md

@@ -62,9 +62,9 @@
 
 ## I/Os
 
-* Python GCSIO is now implemented with GCP GCS Client instead of apitools ([#25676](https://github.com/apache/beam/issues/25676))
 * Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 * TextIO now supports skipping multiple header lines (Java) ([#17990](https://github.com/apache/beam/issues/17990)).
+* Python GCSIO is now implemented with GCP GCS Client instead of apitools ([#25676](https://github.com/apache/beam/issues/25676))
 
 ## New Features / Improvements
 

sdks/python/apache_beam/io/gcp/gcsio.py

@@ -41,6 +41,7 @@
 from google.cloud.storage.fileio import BlobWriter
 
 from apache_beam.internal.gcp import auth
+from apache_beam.options.pipeline_options import GoogleCloudOptions
 from apache_beam.options.pipeline_options import PipelineOptions
 from apache_beam.utils import retry
 from apache_beam.utils.annotations import deprecated
@@ -109,7 +110,8 @@ def __init__(self, storage_client=None, pipeline_options=None):
       credentials = auth.get_service_credentials(pipeline_options)
       if credentials:
         storage_client = storage.Client(
-            credentials=credentials.get_google_auth_credentials())
+            credentials=credentials.get_google_auth_credentials(),
+            project=pipeline_options.view_as(GoogleCloudOptions).project)
       else:
         storage_client = storage.Client.create_anonymous_client()
     self.client = storage_client

sdks/python/apache_beam/runners/dataflow/internal/apiclient.py

@@ -500,7 +500,6 @@ def __init__(self, options, root_staging_location=None):
     else:
       credentials = get_service_credentials(options)
       storage_credentials = credentials.get_google_auth_credentials()
-      storage_project = self.google_cloud_options.project
 
     http_client = get_new_http()
     self._client = dataflow.DataflowV1b3(
@@ -509,9 +508,13 @@ def __init__(self, options, root_staging_location=None):
         get_credentials=(not self.google_cloud_options.no_auth),
         http=http_client,
         response_encoding=get_response_encoding())
-    if storage_credentials and storage_project:
+    if storage_credentials:
+      # Here we explicitly set the project to the value specified in pipeline
+      # options, so the new storage client will be consistent with the previous
+      # client in terms of which GCP project to use.
       self._storage_client = storage.Client(
-          credentials=storage_credentials, project=storage_project)
+          credentials=storage_credentials,
+          project=self.google_cloud_options.project)
     else:
       self._storage_client = storage.Client.create_anonymous_client()
     self._sdk_image_overrides = self._get_sdk_image_overrides(options)

sdks/python/apache_beam/runners/interactive/utils.py

@@ -454,7 +454,10 @@ def assert_bucket_exists(bucket_name):
     from google.cloud import storage
     credentials = auth.get_service_credentials(PipelineOptions())
     if credentials:
-      storage_client = storage.Client(credentials=credentials)
+      # We set project to None, so it will not try to use project id from
+      # the environment (ADC).
+      storage_client = storage.Client(
+          credentials=credentials.get_google_auth_credentials(), project=None)
     else:
       storage_client = storage.Client.create_anonymous_client()
     storage_client.get_bucket(bucket_name)

sdks/python/apache_beam/runners/portability/sdk_container_builder.py

@@ -212,7 +212,8 @@ def __init__(self, options):
     from google.cloud import storage
     if credentials:
       self._storage_client = storage.Client(
-          credentials=credentials.get_google_auth_credentials())
+          credentials=credentials.get_google_auth_credentials(),
+          project=self._google_cloud_options.project)
     else:
       self._storage_client = storage.Client.create_anonymous_client()
     self._cloudbuild_client = cloudbuild.CloudbuildV1(