[Bug]: Failed to run wordcount example after upgrading google-cloud-bigquery to 3.42.0+

[shunping]

What happened?

To reproduce, start a fresh virtual environment and install google-cloud-bigquery>=3.42.0 and apache_beam.

pip install 'apache_beam[gcp]==2.74'

python -m apache_beam.examples.wordcount --output a.txt

INFO:root:Missing pipeline option (runner). Executing pipeline using the default runner: DirectRunner.
Traceback (most recent call last):
  File "<frozen runpy>", line 198, in _run_module_as_main
  File "<frozen runpy>", line 88, in _run_code
  File "/Users/shunping/Projects/v2/venv-2.75rc1/lib/python3.12/site-packages/apache_beam/examples/wordcount.py", line 116, in <module>
    run()
  File "/Users/shunping/Projects/v2/venv-2.75rc1/lib/python3.12/site-packages/apache_beam/examples/wordcount.py", line 90, in run
    lines = pipeline | 'Read' >> ReadFromText(known_args.input)
                                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/shunping/Projects/v2/venv-2.75rc1/lib/python3.12/site-packages/apache_beam/io/textio.py", line 807, in __init__
    self._source = self._source_class(
                   ^^^^^^^^^^^^^^^^^^^
  File "/Users/shunping/Projects/v2/venv-2.75rc1/lib/python3.12/site-packages/apache_beam/io/textio.py", line 143, in __init__
    super().__init__(
  File "/Users/shunping/Projects/v2/venv-2.75rc1/lib/python3.12/site-packages/apache_beam/io/filebasedsource.py", line 126, in __init__
    self._validate()
  File "/Users/shunping/Projects/v2/venv-2.75rc1/lib/python3.12/site-packages/apache_beam/options/value_provider.py", line 192, in _f
    return fnc(self, *args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/shunping/Projects/v2/venv-2.75rc1/lib/python3.12/site-packages/apache_beam/io/filebasedsource.py", line 189, in _validate
    match_result = FileSystems.match([pattern], limits=[1])[0]
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/shunping/Projects/v2/venv-2.75rc1/lib/python3.12/site-packages/apache_beam/io/filesystems.py", line 240, in match
    return filesystem.match(patterns, limits)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/shunping/Projects/v2/venv-2.75rc1/lib/python3.12/site-packages/apache_beam/io/filesystem.py", line 777, in match
    raise BeamIOError("Match operation failed", exceptions)
apache_beam.io.filesystem.BeamIOError: Match operation failed with exceptions {'gs://dataflow-samples/shakespeare/kinglear.txt': ValueError('Context has already been used to create a Connection, it cannot be mutated again')}

This happens when environment variable CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE is set to true. If we set it to false (e.g. CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE=false python -m apache_beam.examples.wordcount --output e.txt), the pipeline can run successfully.

Issue Priority

Priority: 2 (default / most bugs 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

[shunping]

I am able to reproduce the above failure on a freshly installed beam 2.74.0 or 2.75.0rc1. However, the pipeline runs fine on beam 2.73.0 (even with CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE=true).

I recorded pip freeze for both 2.73.0 and 2.74.0, and the diffs are as follows:

$ diff freeze-2.73.txt freeze-2.74.txt
7c7
< apache-beam==2.73.0
---
> apache-beam==2.74.0
45c45
< google-cloud-storage==2.19.0
---
> google-cloud-storage==3.12.0

The version upgrade of google-cloud-storage seems to be relevant to the failure.

[shunping]

Here is a step-by-step summary of how the exception is raised for the wordcount example.

1. Pipeline Execution & GCS I/O
   When the Beam Python Wordcount pipeline is run, it reads from GCS, Beam's ReadFromText transform attempts to read data in chunks from the GCS file stream handled by apache_beam/io/gcp/gcsio.py.

2. Underlying GCS Blob Download
   gcsio.py calls google.cloud.storage.fileio.BlobReader.read(). To fetch file chunks in 3.x code path, google-cloud-storage executes self._blob.download_as_bytes(...) using HTTP Range requests over an AuthorizedSession provided by google-auth.

3. Automatic mTLS & PyOpenSSL Injection
   If GOOGLE_API_USE_CLIENT_CERTIFICATE is not set to false, google-auth detects local client certificates on the developer machine and configures mutual TLS (mTLS). To do this, google-auth calls: urllib3.contrib.pyopenssl.inject_into_urllib3(). This uses PyOpenSSL as the TLS backend for urllib3 and mounts a shared SSLContext on the urllib3 HTTP connection pool.

4. Context Re-use

   • Chunk 1: The first HTTP range request opens "Connection 1" using the shared SSLContext.
   • Chunk 2: When the second range request is issued (or when a connection in the pool is closed and recreated), urllib3 needs to establish "Connection 2" using the same shared SSLContext from the pool.

5. PyOpenSSL Exception
   Before starting the TLS handshake for "Connection 2", urllib3's _ssl_wrap_socket_and_match_hostname attempts to re-apply verification settings to the context. However, in pyOpenSSL >= 26.2.0, PyOpenSSL strictly forbids mutating an SSLContext after it has already created a connection. Therefore, PyOpenSSL raises: ValueError: Context has already been used to create a Connection, it cannot be mutated again.

[shunping]

google-auth has released a newer version 2.55.2 (https://github.com/googleapis/google-cloud-python/releases/tag/google-auth-v2.55.2) in which pyopenssl is replaced with the standard ssl (googleapis/google-cloud-python@247e2ad). I verified that the problem has been resolved with this latest version.