[Feature Request]: Honor --quota_project_id in Python GCS/Pub/Sub IO and the Java-backed BigQuery STORAGE_WRITE_API path #40555

What would you like to happen?

Follow-up to #37431 / #39203.

#39203 adds a --quota_project_id pipeline option to Python GoogleCloudOptions, which attributes quota/billing of API requests to a chosen GCP project (the x-goog-user-project header). The option lives on the GCP-wide options class, but currently only the Python BigQuery IO transforms honor it (ReadFromBigQuery, ReadAllFromBigQuery, WriteToBigQuery with FILE_LOADS/STREAMING_INSERTS, ReadBigQueryChangeHistory).

It should be honored by the rest of the Python GCP IO so the option means what its name says. The client libraries already support it, so this is plumbing only:

• GCS (apache_beam/io/gcp/gcsio.py): storage.Client(...) accepts client_options={'quota_project_id': ...} (honored by google.cloud.client.Client); resourcemanager_v3.ProjectsClient used by _get_project_number takes client_options too. GcsIO already receives pipeline_options, so the value is available where the client is built.
• Pub/Sub (apache_beam/io/gcp/pubsub.py, runners/direct/transform_evaluator.py): pubsub.PublisherClient(...) / SubscriberClient(...) forward client_options to the GAPIC client, which accepts quota_project_id.
• WriteToBigQuery(method=STORAGE_WRITE_API): this path expands to the Java StorageWriteToBigQuery transform. Java already supports --bigQueryQuotaProjectId (BigQueryOptions), but the Python option never reaches it: PipelineOptionsTranslation.fromProto converts lower_underscore → lowerCamel, so quota_project_id arrives as quotaProjectId, which no Java option reads. Options: have the Python transform/expansion forward it under the Java name, or have Java's BigQueryOptions.getBigQueryQuotaProjectId() fall back to a generic quotaProjectId. (Note: on Dataflow, options reach the Java worker in the legacy {"options": {...}} format without case conversion, so a solution must cover that path as well.)
• Other GCP IOs that build clients (Bigtable, Spanner, …) should be checked the same way.

Until then, the option's help text says it is currently honored by BigQuery IO only.

Issue Priority

Priority: 2 (default / most feature requests should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Prism Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner